Adding Interactive API's for MSQ engine

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

@@ -21,12 +21,16 @@
 
 import com.fasterxml.jackson.databind.Module;
 import com.google.inject.Binder;
+import org.apache.druid.client.indexing.HttpIndexingServiceClient;
+import org.apache.druid.client.indexing.IndexingServiceClient;
 import org.apache.druid.discovery.NodeRole;
 import org.apache.druid.guice.Jerseys;
+import org.apache.druid.guice.LazySingleton;
 import org.apache.druid.guice.LifecycleModule;
 import org.apache.druid.guice.annotations.LoadScope;
 import org.apache.druid.initialization.DruidModule;
-import org.apache.druid.msq.sql.SqlTaskResource;
+import org.apache.druid.msq.sql.resources.SqlStatementResource;
+import org.apache.druid.msq.sql.resources.SqlTaskResource;
 
 import java.util.Collections;
 import java.util.List;
@@ -43,6 +47,9 @@ public void configure(Binder binder)
     // Force eager initialization.
     LifecycleModule.register(binder, SqlTaskResource.class);
     Jerseys.addResource(binder, SqlTaskResource.class);
+    LifecycleModule.register(binder, SqlStatementResource.class);
+    binder.bind(IndexingServiceClient.class).to(HttpIndexingServiceClient.class).in(LazySingleton.class);
+    Jerseys.addResource(binder, SqlStatementResource.class);
   }
 
   @Override

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

@@ -30,6 +30,7 @@
 import com.google.inject.Injector;
 import com.google.inject.Key;
 import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.druid.client.indexing.ClientTaskQuery;
 import org.apache.druid.guice.annotations.EscalatedGlobal;
 import org.apache.druid.indexer.TaskStatus;
 import org.apache.druid.indexing.common.TaskLock;
@@ -49,6 +50,7 @@
 import org.apache.druid.rpc.ServiceClientFactory;
 import org.apache.druid.rpc.StandardRetryPolicy;
 import org.apache.druid.rpc.indexing.OverlordClient;
+import org.apache.druid.segment.column.ColumnType;
 import org.apache.druid.server.security.ResourceAction;
 import org.apache.druid.sql.calcite.run.SqlResults;
 import org.joda.time.Interval;
@@ -60,7 +62,7 @@
 import java.util.Set;
 
 @JsonTypeName(MSQControllerTask.TYPE)
-public class MSQControllerTask extends AbstractTask
+public class MSQControllerTask extends AbstractTask implements ClientTaskQuery
 {
   public static final String TYPE = "query_controller";
   public static final String DUMMY_DATASOURCE_FOR_SELECT = "__query_select";
@@ -91,6 +93,9 @@ public class MSQControllerTask extends AbstractTask
   @Nullable
   private final List<SqlTypeName> sqlTypeNames;
 
+  @Nullable
+  private final List<ColumnType> nativeTypeNames;
+
   // Using an Injector directly because tasks do not have a way to provide their own Guice modules.
   @JacksonInject
   private Injector injector;
@@ -105,6 +110,7 @@ public MSQControllerTask(
       @JsonProperty("sqlQueryContext") @Nullable Map<String, Object> sqlQueryContext,
       @JsonProperty("sqlResultsContext") @Nullable SqlResults.Context sqlResultsContext,
       @JsonProperty("sqlTypeNames") @Nullable List<SqlTypeName> sqlTypeNames,
+      @JsonProperty("nativeTypeNames") @Nullable List<ColumnType> nativeTypeNames,
       @JsonProperty("context") @Nullable Map<String, Object> context
   )
   {
@@ -121,6 +127,7 @@ public MSQControllerTask(
     this.sqlQueryContext = sqlQueryContext;
     this.sqlResultsContext = sqlResultsContext;
     this.sqlTypeNames = sqlTypeNames;
+    this.nativeTypeNames = nativeTypeNames;
 
     addToContext(Tasks.FORCE_TIME_CHUNK_LOCK_KEY, true);
   }
@@ -154,6 +161,15 @@ public List<SqlTypeName> getSqlTypeNames()
     return sqlTypeNames;
   }
 
+
+  @Nullable
+  @JsonProperty
+  @JsonInclude(JsonInclude.Include.NON_NULL)
+  public List<ColumnType> getNativeTypeNames()
+  {
+    return nativeTypeNames;
+  }
+
   @Nullable
   @JsonProperty
   @JsonInclude(JsonInclude.Include.NON_NULL)

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/error/InsertCannotBeEmptyFault.java

@@ -29,7 +29,7 @@
 @JsonTypeName(InsertCannotBeEmptyFault.CODE)
 public class InsertCannotBeEmptyFault extends BaseMSQFault
 {
-  static final String CODE = "InsertCannotBeEmpty";
+  public static final String CODE = "InsertCannotBeEmpty";
 
   private final String dataSource;
 

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

@@ -78,6 +78,8 @@ public class MSQTaskQueryMaker implements QueryMaker
   private static final String DESTINATION_DATASOURCE = "dataSource";
   private static final String DESTINATION_REPORT = "taskReport";
 
+  public static final String USER_KEY = "__user";
+
   private static final Granularity DEFAULT_SEGMENT_GRANULARITY = Granularities.ALL;
 
   private final String targetDataSource;
@@ -116,6 +118,9 @@ public QueryResponse<Object[]> runQuery(final DruidQuery druidQuery)
     // Native query context: sqlQueryContext plus things that we add prior to creating a controller task.
     final Map<String, Object> nativeQueryContext = new HashMap<>(sqlQueryContext.asMap());
 
+    // adding user
+    nativeQueryContext.put(USER_KEY, plannerContext.getAuthenticationResult().getIdentity());
+
     final String msqMode = MultiStageQueryContext.getMSQMode(sqlQueryContext);
     if (msqMode != null) {
       MSQMode.populateDefaultQueryContext(msqMode, nativeQueryContext);
@@ -174,6 +179,7 @@ public QueryResponse<Object[]> runQuery(final DruidQuery druidQuery)
         finalizeAggregations ? null /* Not needed */ : buildAggregationIntermediateTypeMap(druidQuery);
 
     final List<SqlTypeName> sqlTypeNames = new ArrayList<>();
+    final List<ColumnType> columnTypeList = new ArrayList<>();
     final List<ColumnMapping> columnMappings = QueryUtils.buildColumnMappings(fieldMapping, druidQuery);
 
     for (final Pair<Integer, String> entry : fieldMapping) {
@@ -187,8 +193,8 @@ public QueryResponse<Object[]> runQuery(final DruidQuery druidQuery)
       } else {
         sqlTypeName = druidQuery.getOutputRowType().getFieldList().get(entry.getKey()).getType().getSqlTypeName();
       }
-
       sqlTypeNames.add(sqlTypeName);
+      columnTypeList.add(druidQuery.getOutputRowSignature().getColumnType(queryColumn).orElse(ColumnType.STRING));
     }
 
     final MSQDestination destination;
@@ -248,6 +254,7 @@ public QueryResponse<Object[]> runQuery(final DruidQuery druidQuery)
         plannerContext.queryContextMap(),
         SqlResults.Context.fromPlannerContext(plannerContext),
         sqlTypeNames,
+        columnTypeList,
         null
     );
 

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

@@ -296,7 +296,9 @@ private static void validateLimitAndOffset(final RelNode topRel, final boolean l
   private static RelDataType getMSQStructType(RelDataTypeFactory typeFactory)
   {
     return typeFactory.createStructType(
-        ImmutableList.of(Calcites.createSqlType(typeFactory, SqlTypeName.VARCHAR)),
+        ImmutableList.of(
+            Calcites.createSqlType(typeFactory, SqlTypeName.VARCHAR)
+        ),
         TASK_STRUCT_FIELD_NAMES
     );
   }

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

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.msq.sql;
+
+import org.apache.druid.sql.http.SqlQuery;
+
+import javax.servlet.http.HttpServletRequest;
+
+/**
+ * Represents the status of the sql statements issues via
+ * {@link org.apache.druid.msq.sql.resources.SqlStatementResource#doPost(SqlQuery, HttpServletRequest)} and returned in
+ * {@link org.apache.druid.msq.sql.entity.SqlStatementResult}
+ */
+public enum SqlStatementState
+{
+  // The statement is accepted but not yes assigned any worker. In MSQ engine, the statement is in ACCEPTED state
+  // till the overlord assigns a TaskLocation to the controller task.
+  ACCEPTED,
+  // The statement is running.
+  RUNNING,
+  // The statement is successful.
+  SUCCESS,
+  // The statement failed.
+  FAILED
+}

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

@@ -25,6 +25,7 @@
 import com.google.common.base.Preconditions;
 import org.apache.druid.error.ErrorResponse;
 import org.apache.druid.indexer.TaskState;
+import org.apache.druid.msq.sql.resources.SqlTaskResource;
 
 import javax.annotation.Nullable;
 import java.util.Objects;

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

@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.msq.sql.entity;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Objects;
+
+/**
+ * The column name and its sql {@link org.apache.calcite.sql.type.SqlTypeName} and its native {@link org.apache.druid.segment.column.ColumnType}
+ */
+
+public class ColumnNameAndTypes
+{
+
+  private final String colName;
+  private final String sqlTypeName;
+
+  private final String nativeTypeName;
+
+  @JsonCreator
+  public ColumnNameAndTypes(
+      @JsonProperty("name") String colName,
+      @JsonProperty("type") String sqlTypeName,
+      @JsonProperty("nativeType") String nativeTypeName
+  )
+  {
+
+    this.colName = colName;
+    this.sqlTypeName = sqlTypeName;
+    this.nativeTypeName = nativeTypeName;
+  }
+
+  @JsonProperty("name")
+  public String getColName()
+  {
+    return colName;
+  }
+
+  @JsonProperty("type")
+  public String getSqlTypeName()
+  {
+    return sqlTypeName;
+  }
+
+  @JsonProperty("nativeType")
+  public String getNativeTypeName()
+  {
+    return nativeTypeName;
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    ColumnNameAndTypes that = (ColumnNameAndTypes) o;
+    return Objects.equals(colName, that.colName)
+           && Objects.equals(sqlTypeName, that.sqlTypeName)
+           && Objects.equals(nativeTypeName, that.nativeTypeName);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(colName, sqlTypeName, nativeTypeName);
+  }
+
+  @Override
+  public String toString()
+  {
+    return "ColumnNameAndTypes{" +
+           "colName='" + colName + '\'' +
+           ", sqlTypeName='" + sqlTypeName + '\'' +
+           ", nativeTypeName='" + nativeTypeName + '\'' +
+           '}';
+  }
+}
+
+