MSQ profile for Brokers and Historicals. (apache#17140)

This patch adds a profile of MSQ named "Dart" that runs on Brokers and
Historicals, and which is compatible with the standard SQL query API.
For more high-level description, and notes on future work, refer to apache#17139.

This patch contains the following changes, grouped into packages.

Controller (org.apache.druid.msq.dart.controller):

The controller runs on Brokers. Main classes are,

- DartSqlResource, which serves /druid/v2/sql/dart/.
- DartSqlEngine and DartQueryMaker, the entry points from SQL that actually
  run the MSQ controller code.
- DartControllerContext, which configures the MSQ controller.
- DartMessageRelays, which sets up relays (see "message relays" below) to read
  messages from workers' DartControllerClients.
- DartTableInputSpecSlicer, which assigns work based on a TimelineServerView.

Worker (org.apache.druid.msq.dart.worker)

The worker runs on Historicals. Main classes are,

- DartWorkerResource, which supplies the regular MSQ WorkerResource, plus
  Dart-specific APIs.
- DartWorkerRunner, which runs MSQ worker code.
- DartWorkerContext, which configures the MSQ worker.
- DartProcessingBuffersProvider, which provides processing buffers from
  sliced-up merge buffers.
- DartDataSegmentProvider, which provides segments from the Historical's
  local cache.

Message relays (org.apache.druid.messages):

To avoid the need for Historicals to contact Brokers during a query, which
would create opportunities for queries to get stuck, all connections are
opened from Broker to Historical. This is made possible by a message relay
system, where the relay server (worker) has an outbox of messages.

The relay client (controller) connects to the outbox and retrieves messages.
Code for this system lives in the "server" package to keep it separate from
the MSQ extension and make it easier to maintain. The worker-to-controller
ControllerClient is implemented using message relays.

Other changes:

- Controller: Added the method "hasWorker". Used by the ControllerMessageListener
  to notify the appropriate controllers when a worker fails.
- WorkerResource: No longer tries to respond more than once in the
  "httpGetChannelData" API. This comes up when a response due to resolved future
  is ready at about the same time as a timeout occurs.
- MSQTaskQueryMaker: Refactor to separate out some useful functions for reuse
  in DartQueryMaker.
- SqlEngine: Add "queryContext" to "resultTypeForSelect" and "resultTypeForInsert".
  This allows the DartSqlEngine to modify result format based on whether a "fullReport"
  context parameter is set.
- LimitedOutputStream: New utility class. Used when in "fullReport" mode.
- TimelineServerView: Add getDruidServerMetadata as a performance optimization.
- CliHistorical: Add SegmentWrangler, so it can query inline data, lookups, etc.
- ServiceLocation: Add "fromUri" method, relocating some code from ServiceClientImpl.
- FixedServiceLocator: New locator for a fixed set of service locations. Useful for
  URI locations.

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/dart/Dart.java

@@ -0,0 +1,37 @@
+/*
+ * 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.dart;
+
+import com.google.inject.BindingAnnotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Binding annotation for implements of interfaces that are Dart (MSQ-on-Broker-and-Historicals) focused.
+ */
+@Target({ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@BindingAnnotation
+public @interface Dart
+{
+}

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/dart/DartResourcePermissionMapper.java

@@ -0,0 +1,57 @@
+/*
+ * 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.dart;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.druid.msq.dart.controller.http.DartSqlResource;
+import org.apache.druid.msq.dart.worker.http.DartWorkerResource;
+import org.apache.druid.msq.rpc.ResourcePermissionMapper;
+import org.apache.druid.msq.rpc.WorkerResource;
+import org.apache.druid.server.security.Action;
+import org.apache.druid.server.security.Resource;
+import org.apache.druid.server.security.ResourceAction;
+
+import java.util.List;
+
+public class DartResourcePermissionMapper implements ResourcePermissionMapper
+{
+  /**
+   * Permissions for admin APIs in {@link DartWorkerResource} and {@link WorkerResource}. Note that queries from
+   * end users go through {@link DartSqlResource}, which wouldn't use these mappings.
+   */
+  @Override
+  public List<ResourceAction> getAdminPermissions()
+  {
+    return ImmutableList.of(
+        new ResourceAction(Resource.STATE_RESOURCE, Action.READ),
+        new ResourceAction(Resource.STATE_RESOURCE, Action.WRITE)
+    );
+  }
+
+  /**
+   * Permissions for per-query APIs in {@link DartWorkerResource} and {@link WorkerResource}. Note that queries from
+   * end users go through {@link DartSqlResource}, which wouldn't use these mappings.
+   */
+  @Override
+  public List<ResourceAction> getQueryPermissions(String queryId)
+  {
+    return getAdminPermissions();
+  }
+}

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/dart/controller/ControllerHolder.java

@@ -0,0 +1,166 @@
+/*
+ * 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.dart.controller;
+
+import com.google.common.base.Preconditions;
+import org.apache.druid.msq.dart.worker.DartWorkerClient;
+import org.apache.druid.msq.dart.worker.WorkerId;
+import org.apache.druid.msq.exec.Controller;
+import org.apache.druid.msq.exec.ControllerContext;
+import org.apache.druid.msq.exec.QueryListener;
+import org.apache.druid.msq.indexing.error.MSQErrorReport;
+import org.apache.druid.msq.indexing.error.WorkerFailedFault;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.joda.time.DateTime;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Holder for {@link Controller}, stored in {@link DartControllerRegistry}.
+ */
+public class ControllerHolder
+{
+  public enum State
+  {
+    /**
+     * Query has been accepted, but not yet {@link Controller#run(QueryListener)}.
+     */
+    ACCEPTED,
+
+    /**
+     * Query has had {@link Controller#run(QueryListener)} called.
+     */
+    RUNNING,
+
+    /**
+     * Query has been canceled.
+     */
+    CANCELED
+  }
+
+  private final Controller controller;
+  private final ControllerContext controllerContext;
+  private final String sqlQueryId;
+  private final String sql;
+  private final AuthenticationResult authenticationResult;
+  private final DateTime startTime;
+  private final AtomicReference<State> state = new AtomicReference<>(State.ACCEPTED);
+
+  public ControllerHolder(
+      final Controller controller,
+      final ControllerContext controllerContext,
+      final String sqlQueryId,
+      final String sql,
+      final AuthenticationResult authenticationResult,
+      final DateTime startTime
+  )
+  {
+    this.controller = Preconditions.checkNotNull(controller, "controller");
+    this.controllerContext = controllerContext;
+    this.sqlQueryId = Preconditions.checkNotNull(sqlQueryId, "sqlQueryId");
+    this.sql = sql;
+    this.authenticationResult = authenticationResult;
+    this.startTime = Preconditions.checkNotNull(startTime, "startTime");
+  }
+
+  public Controller getController()
+  {
+    return controller;
+  }
+
+  public String getSqlQueryId()
+  {
+    return sqlQueryId;
+  }
+
+  public String getSql()
+  {
+    return sql;
+  }
+
+  public AuthenticationResult getAuthenticationResult()
+  {
+    return authenticationResult;
+  }
+
+  public DateTime getStartTime()
+  {
+    return startTime;
+  }
+
+  public State getState()
+  {
+    return state.get();
+  }
+
+  /**
+   * Call when a worker has gone offline. Closes its client and sends a {@link Controller#workerError}
+   * to the controller.
+   */
+  public void workerOffline(final WorkerId workerId)
+  {
+    final String workerIdString = workerId.toString();
+
+    if (controllerContext instanceof DartControllerContext) {
+      // For DartControllerContext, newWorkerClient() returns the same instance every time.
+      // This will always be DartControllerContext in production; the instanceof check is here because certain
+      // tests use a different context class.
+      ((DartWorkerClient) controllerContext.newWorkerClient()).closeClient(workerId.getHostAndPort());
+    }
+
+    if (controller.hasWorker(workerIdString)) {
+      controller.workerError(
+          MSQErrorReport.fromFault(
+              workerIdString,
+              workerId.getHostAndPort(),
+              null,
+              new WorkerFailedFault(workerIdString, "Worker went offline")
+          )
+      );
+    }
+  }
+
+  /**
+   * Places this holder into {@link State#CANCELED}. Calls {@link Controller#stop()} if it was previously in
+   * state {@link State#RUNNING}.
+   */
+  public void cancel()
+  {
+    if (state.getAndSet(State.CANCELED) == State.RUNNING) {
+      controller.stop();
+    }
+  }
+
+  /**
+   * Calls {@link Controller#run(QueryListener)}, and returns true, if this holder was previously in state
+   * {@link State#ACCEPTED}. Otherwise returns false.
+   *
+   * @return whether {@link Controller#run(QueryListener)} was called.
+   */
+  public boolean run(final QueryListener listener) throws Exception
+  {
+    if (state.compareAndSet(State.ACCEPTED, State.RUNNING)) {
+      controller.run(listener);
+      return true;
+    } else {
+      return false;
+    }
+  }
+}

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/dart/controller/ControllerMessageListener.java

@@ -0,0 +1,68 @@
+/*
+ * 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.dart.controller;
+
+import com.google.inject.Inject;
+import org.apache.druid.messages.client.MessageListener;
+import org.apache.druid.msq.dart.controller.messages.ControllerMessage;
+import org.apache.druid.msq.dart.worker.WorkerId;
+import org.apache.druid.msq.exec.Controller;
+import org.apache.druid.msq.indexing.error.MSQErrorReport;
+import org.apache.druid.server.DruidNode;
+
+/**
+ * Listener for worker-to-controller messages.
+ * Also responsible for calling {@link Controller#workerError(MSQErrorReport)} when a worker server goes away.
+ */
+public class ControllerMessageListener implements MessageListener<ControllerMessage>
+{
+  private final DartControllerRegistry controllerRegistry;
+
+  @Inject
+  public ControllerMessageListener(final DartControllerRegistry controllerRegistry)
+  {
+    this.controllerRegistry = controllerRegistry;
+  }
+
+  @Override
+  public void messageReceived(ControllerMessage message)
+  {
+    final ControllerHolder holder = controllerRegistry.get(message.getQueryId());
+    if (holder != null) {
+      message.handle(holder.getController());
+    }
+  }
+
+  @Override
+  public void serverAdded(DruidNode node)
+  {
+    // Nothing to do.
+  }
+
+  @Override
+  public void serverRemoved(DruidNode node)
+  {
+    for (final ControllerHolder holder : controllerRegistry.getAllHolders()) {
+      final Controller controller = holder.getController();
+      final WorkerId workerId = WorkerId.fromDruidNode(node, controller.queryId());
+      holder.workerOffline(workerId);
+    }
+  }
+}