MSQ profile for Brokers and Historicals.

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,46 @@
+/*
+ * 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.rpc.ResourcePermissionMapper;
+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
+{
+  @Override
+  public List<ResourceAction> getAdminPermissions()
+  {
+    return ImmutableList.of(
+        new ResourceAction(Resource.STATE_RESOURCE, Action.READ),
+        new ResourceAction(Resource.STATE_RESOURCE, Action.WRITE)
+    );
+  }
+
+  @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,81 @@
+/*
+ * 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.exec.Controller;
+import org.apache.druid.msq.util.MSQTaskQueryMakerUtils;
+import org.joda.time.DateTime;
+
+import javax.annotation.Nullable;
+
+/**
+ * Holder for {@link Controller}, stored in {@link DartControllerRegistry}.
+ */
+public class ControllerHolder
+{
+  private final Controller controller;
+  private final String sqlQueryId;
+  private final String sql;
+  @Nullable
+  private final String identity;
+  private final DateTime startTime;
+
+  public ControllerHolder(
+      final Controller controller,
+      final String sqlQueryId,
+      final String sql,
+      final String identity,
+      final DateTime startTime
+  )
+  {
+    this.controller = Preconditions.checkNotNull(controller, "controller");
+    this.sqlQueryId = Preconditions.checkNotNull(sqlQueryId, "sqlQueryId");
+    this.sql = MSQTaskQueryMakerUtils.maskSensitiveJsonKeys(Preconditions.checkNotNull(sql, "sql"));
+    this.identity = identity;
+    this.startTime = Preconditions.checkNotNull(startTime, "startTime");
+  }
+
+  public Controller getController()
+  {
+    return controller;
+  }
+
+  public String getSqlQueryId()
+  {
+    return sqlQueryId;
+  }
+
+  public String getSql()
+  {
+    return sql;
+  }
+
+  @Nullable
+  public String getIdentity()
+  {
+    return identity;
+  }
+
+  public DateTime getStartTime()
+  {
+    return startTime;
+  }
+}

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

@@ -0,0 +1,85 @@
+/*
+ * 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.msq.indexing.error.WorkerFailedFault;
+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 serverAdded(DruidNode node)
+  {
+    // Fail workers when they're added, because when they're added, they shouldn't be running anything. If they are
+    // running something, cancel it.
+    workerFailed(node);
+  }
+
+  @Override
+  public void serverRemoved(DruidNode node)
+  {
+    workerFailed(node);
+  }
+
+  @Override
+  public void messageReceived(ControllerMessage message)
+  {
+    final Controller controller = controllerRegistry.get(message.getQueryId());
+    if (controller != null) {
+      message.handle(controller);
+    }
+  }
+
+  private void workerFailed(final DruidNode node)
+  {
+    for (final ControllerHolder holder : controllerRegistry.getAllHolders()) {
+      final Controller controller = holder.getController();
+      final String workerId = WorkerId.fromDruidNode(node, controller.queryId()).toString();
+      if (controller.hasWorker(workerId)) {
+        controller.workerError(
+            MSQErrorReport.fromFault(
+                workerId,
+                node.getHost(),
+                null,
+                new WorkerFailedFault(workerId, "Worker went offline")
+            )
+        );
+      }
+    }
+  }
+}

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

@@ -0,0 +1,116 @@
+/*
+ * 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.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonValue;
+import com.google.common.base.Preconditions;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.msq.exec.Controller;
+import org.apache.druid.server.DruidNode;
+
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Unique identifier for a particular run of the controller server. This does not correspond to any specific
+ * {@link Controller}; all controllers on the same server use the same ID.
+ */
+public class ControllerServerId
+{
+  private static final Pattern PATTERN = Pattern.compile("^(.+):(\\d+)$");
+
+  private final String host;
+  private final long epoch;
+  private final String fullString;
+
+  public ControllerServerId(final String host, final long epoch)
+  {
+    this.host = Preconditions.checkNotNull(host, "host");
+    this.epoch = epoch;
+    this.fullString = host + ':' + epoch;
+  }
+
+  @JsonCreator
+  public static ControllerServerId fromString(final String s)
+  {
+    final Matcher matcher = PATTERN.matcher(s);
+    if (matcher.matches()) {
+      return new ControllerServerId(matcher.group(1), Long.parseLong(matcher.group(2)));
+    } else {
+      throw new IAE("Invalid controllerId[%s]", s);
+    }
+  }
+
+  /**
+   * Host and port, from {@link DruidNode#getHostAndPortToUse()}, of the controller server.
+   */
+  @JsonProperty
+  public String getHost()
+  {
+    return host;
+  }
+
+  /**
+   * Epoch of the controller server. Increased every time the server reboots. This can be used to determine if
+   * the controller server being communicated with is the "same" controller server as one we know about with the
+   * same hostname.
+   */
+  @JsonProperty
+  public long getEpoch()
+  {
+    return epoch;
+  }
+
+  /**
+   * Returns whether this controllerId replaces another one, i.e., if the host is the same and epoch is greater.
+   */
+  public boolean replaces(final ControllerServerId otherId)
+  {
+    return otherId.getHost().equals(host) && epoch > otherId.getEpoch();
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    ControllerServerId that = (ControllerServerId) o;
+    return fullString.equals(that.fullString);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return fullString.hashCode();
+  }
+
+  @Override
+  @JsonValue
+  public String toString()
+  {
+    return fullString;
+  }
+}