You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ja...@apache.org on 2016/01/11 07:44:49 UTC
[1/3] drill git commit: Fix issue where FutureBitComment didn't
return buffer as part of future.
Repository: drill
Updated Branches:
refs/heads/master 5cb8eb95e -> f964908ae
Fix issue where FutureBitComment didn't return buffer as part of future.
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/467c405c
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/467c405c
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/467c405c
Branch: refs/heads/master
Commit: 467c405cf48017f668f779d24978b4ffaa4ea8f0
Parents: 5cb8eb9
Author: Jacques Nadeau <ja...@apache.org>
Authored: Sat Jan 2 16:32:36 2016 -0800
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Jan 9 18:33:06 2016 -0800
----------------------------------------------------------------------
.../java/org/apache/drill/exec/rpc/FutureBitCommand.java | 1 +
.../java/org/apache/drill/exec/rpc/RpcCheckedFuture.java | 9 ++++++---
2 files changed, 7 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/467c405c/exec/rpc/src/main/java/org/apache/drill/exec/rpc/FutureBitCommand.java
----------------------------------------------------------------------
diff --git a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/FutureBitCommand.java b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/FutureBitCommand.java
index 53d0772..1b19561 100644
--- a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/FutureBitCommand.java
+++ b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/FutureBitCommand.java
@@ -55,6 +55,7 @@ public abstract class FutureBitCommand<T extends MessageLite, C extends RemoteCo
@Override
public void success(T value, ByteBuf buf) {
+ parentFuture.setBuffer(buf);
settableFuture.set(value);
}
http://git-wip-us.apache.org/repos/asf/drill/blob/467c405c/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RpcCheckedFuture.java
----------------------------------------------------------------------
diff --git a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RpcCheckedFuture.java b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RpcCheckedFuture.java
index 11b07ad..8746a20 100644
--- a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RpcCheckedFuture.java
+++ b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RpcCheckedFuture.java
@@ -30,8 +30,11 @@ public class RpcCheckedFuture<T> extends AbstractCheckedFuture<T, RpcException>
super(delegate);
}
- public void set(T obj, ByteBuf buffer){
- this.buffer = buffer;
+ public void setBuffer(ByteBuf buffer) {
+ if (buffer != null) {
+ buffer.retain();
+ this.buffer = buffer;
+ }
}
@Override
@@ -41,7 +44,7 @@ public class RpcCheckedFuture<T> extends AbstractCheckedFuture<T, RpcException>
@Override
public ByteBuf getBuffer() {
- return null;
+ return buffer;
}
}
[2/3] drill git commit: DRILL-4238: Add a custom RPC interface on the
Control channel for extensible communication between bits.
Posted by ja...@apache.org.
DRILL-4238: Add a custom RPC interface on the Control channel for extensible communication between bits.
This closes #313.
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/67d5cc6d
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/67d5cc6d
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/67d5cc6d
Branch: refs/heads/master
Commit: 67d5cc6df0ca394dd26ff5c0a2b0510779bad949
Parents: 467c405
Author: Jacques Nadeau <ja...@apache.org>
Authored: Sat Jan 2 14:55:21 2016 -0800
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sun Jan 10 22:41:50 2016 -0800
----------------------------------------------------------------------
.../exec/rpc/control/ControlRpcConfig.java | 2 +
.../drill/exec/rpc/control/ControlTunnel.java | 174 +++++-
.../drill/exec/rpc/control/Controller.java | 63 ++
.../drill/exec/rpc/control/ControllerImpl.java | 13 +
.../exec/rpc/control/CustomHandlerRegistry.java | 124 ++++
.../rpc/control/DefaultInstanceHandler.java | 4 +-
.../exec/work/batch/ControlMessageHandler.java | 14 +
.../exec/rpc/control/TestCustomTunnel.java | 138 +++++
.../org/apache/drill/exec/proto/BitControl.java | 599 +++++++++++++++++--
.../drill/exec/proto/SchemaBitControl.java | 119 ++++
.../drill/exec/proto/beans/CustomMessage.java | 186 ++++++
protocol/src/main/protobuf/BitControl.proto | 7 +
12 files changed, 1397 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/67d5cc6d/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControlRpcConfig.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControlRpcConfig.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControlRpcConfig.java
index 44046c9..ec09a98 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControlRpcConfig.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControlRpcConfig.java
@@ -22,6 +22,7 @@ import java.util.concurrent.Executor;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.proto.BitControl.BitControlHandshake;
+import org.apache.drill.exec.proto.BitControl.CustomMessage;
import org.apache.drill.exec.proto.BitControl.FinishedReceiver;
import org.apache.drill.exec.proto.BitControl.FragmentStatus;
import org.apache.drill.exec.proto.BitControl.InitializeFragments;
@@ -50,6 +51,7 @@ public class ControlRpcConfig {
.add(RpcType.REQ_FRAGMENT_STATUS, FragmentStatus.class, RpcType.ACK, Ack.class)
.add(RpcType.REQ_QUERY_STATUS, QueryId.class, RpcType.RESP_QUERY_STATUS, QueryProfile.class)
.add(RpcType.REQ_UNPAUSE_FRAGMENT, FragmentHandle.class, RpcType.ACK, Ack.class)
+ .add(RpcType.REQ_CUSTOM, CustomMessage.class, RpcType.RESP_CUSTOM, CustomMessage.class)
.build();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/67d5cc6d/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControlTunnel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControlTunnel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControlTunnel.java
index b90912d..ff8be1d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControlTunnel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControlTunnel.java
@@ -17,6 +17,13 @@
*/
package org.apache.drill.exec.rpc.control;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.DrillBuf;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.drill.exec.proto.BitControl.CustomMessage;
import org.apache.drill.exec.proto.BitControl.FinishedReceiver;
import org.apache.drill.exec.proto.BitControl.FragmentStatus;
import org.apache.drill.exec.proto.BitControl.InitializeFragments;
@@ -26,12 +33,16 @@ import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
import org.apache.drill.exec.proto.UserBitShared.QueryId;
import org.apache.drill.exec.proto.UserBitShared.QueryProfile;
-import org.apache.drill.exec.proto.helper.QueryIdHelper;
import org.apache.drill.exec.rpc.DrillRpcFuture;
import org.apache.drill.exec.rpc.FutureBitCommand;
import org.apache.drill.exec.rpc.ListeningCommand;
+import org.apache.drill.exec.rpc.RpcException;
import org.apache.drill.exec.rpc.RpcOutcomeListener;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Message;
+import com.google.protobuf.Parser;
+
public class ControlTunnel {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ControlTunnel.class);
@@ -176,4 +187,165 @@ public class ControlTunnel {
connection.send(outcomeListener, RpcType.REQ_QUERY_CANCEL, queryId, Ack.class);
}
}
+
+ public <SEND extends Message, RECEIVE extends Message> CustomTunnel<SEND, RECEIVE> getCustomTunnel(
+ int messageTypeId, Class<SEND> clazz, Parser<RECEIVE> parser) {
+ return new CustomTunnel<SEND, RECEIVE>(messageTypeId, parser);
+ }
+
+ private static class CustomMessageSender extends ListeningCommand<CustomMessage, ControlConnection> {
+
+ private CustomMessage message;
+ private ByteBuf[] dataBodies;
+
+ public CustomMessageSender(RpcOutcomeListener<CustomMessage> listener, CustomMessage message, ByteBuf[] dataBodies) {
+ super(listener);
+ this.message = message;
+ this.dataBodies = dataBodies;
+ }
+
+ @Override
+ public void doRpcCall(RpcOutcomeListener<CustomMessage> outcomeListener, ControlConnection connection) {
+ connection.send(outcomeListener, RpcType.REQ_CUSTOM, message, CustomMessage.class, dataBodies);
+ }
+
+ }
+
+ private static class SyncCustomMessageSender extends FutureBitCommand<CustomMessage, ControlConnection> {
+
+ private CustomMessage message;
+ private ByteBuf[] dataBodies;
+
+ public SyncCustomMessageSender(CustomMessage message, ByteBuf[] dataBodies) {
+ super();
+ this.message = message;
+ this.dataBodies = dataBodies;
+ }
+
+ @Override
+ public void doRpcCall(RpcOutcomeListener<CustomMessage> outcomeListener, ControlConnection connection) {
+ connection.send(outcomeListener, RpcType.REQ_CUSTOM, message, CustomMessage.class, dataBodies);
+ }
+ }
+
+ /**
+ * A class used to return a synchronous future when doing custom rpc messages.
+ * @param <RECEIVE>
+ * The type of message that will be returned.
+ */
+ public class CustomFuture<RECEIVE> {
+
+ private Parser<RECEIVE> parser;
+ private DrillRpcFuture<CustomMessage> future;
+
+ public CustomFuture(Parser<RECEIVE> parser, DrillRpcFuture<CustomMessage> future) {
+ super();
+ this.parser = parser;
+ this.future = future;
+ }
+
+ public RECEIVE get() throws RpcException, InvalidProtocolBufferException {
+ CustomMessage message = future.checkedGet();
+ return parser.parseFrom(message.getMessage());
+ }
+
+ public RECEIVE get(long timeout, TimeUnit unit) throws RpcException, TimeoutException,
+ InvalidProtocolBufferException {
+ CustomMessage message = future.checkedGet(timeout, unit);
+ return parser.parseFrom(message.getMessage());
+ }
+
+ public DrillBuf getBuffer() throws RpcException {
+ return (DrillBuf) future.getBuffer();
+ }
+
+ }
+
+ /**
+ * A special tunnel that can be used for custom types of messages. Its lifecycle is tied to the underlying
+ * ControlTunnel.
+ * @param <SEND>
+ * The type of message the control tunnel will be able to send.
+ * @param <RECEIVE>
+ * The expected response the control tunnel expects to receive.
+ */
+ public class CustomTunnel<SEND extends Message, RECEIVE extends Message> {
+ private int messageTypeId;
+ private Parser<RECEIVE> parser;
+
+ private CustomTunnel(int messageTypeId, Parser<RECEIVE> parser) {
+ super();
+ this.messageTypeId = messageTypeId;
+ this.parser = parser;
+ }
+
+ /**
+ * Send a message and receive a future for monitoring the outcome.
+ * @param messageToSend
+ * The structured message to send.
+ * @param dataBodies
+ * One or more optional unstructured messages to append to the structure message.
+ * @return The CustomFuture that can be used to wait for the response.
+ */
+ public CustomFuture<RECEIVE> send(SEND messageToSend, ByteBuf... dataBodies) {
+ final CustomMessage customMessage = CustomMessage.newBuilder()
+ .setMessage(messageToSend.toByteString())
+ .setType(messageTypeId)
+ .build();
+ final SyncCustomMessageSender b = new SyncCustomMessageSender(customMessage, dataBodies);
+ manager.runCommand(b);
+ DrillRpcFuture<CustomMessage> innerFuture = b.getFuture();
+ return new CustomFuture<RECEIVE>(parser, innerFuture);
+ }
+
+ /**
+ * Send a message using a custom listener.
+ * @param listener
+ * The listener to inform of the outcome of the sent message.
+ * @param messageToSend
+ * The structured message to send.
+ * @param dataBodies
+ * One or more optional unstructured messages to append to the structure message.
+ */
+ public void send(RpcOutcomeListener<RECEIVE> listener, SEND messageToSend, ByteBuf... dataBodies) {
+ final CustomMessage customMessage = CustomMessage.newBuilder()
+ .setMessage(messageToSend.toByteString())
+ .setType(messageTypeId)
+ .build();
+ manager.runCommand(new CustomMessageSender(new CustomTunnelListener(listener), customMessage, dataBodies));
+ }
+
+ private class CustomTunnelListener implements RpcOutcomeListener<CustomMessage> {
+ final RpcOutcomeListener<RECEIVE> innerListener;
+
+ public CustomTunnelListener(RpcOutcomeListener<RECEIVE> innerListener) {
+ super();
+ this.innerListener = innerListener;
+ }
+
+ @Override
+ public void failed(RpcException ex) {
+ innerListener.failed(ex);
+ }
+
+ @Override
+ public void success(CustomMessage value, ByteBuf buffer) {
+ try {
+ RECEIVE message = parser.parseFrom(value.getMessage());
+ innerListener.success(message, buffer);
+ } catch (InvalidProtocolBufferException e) {
+ innerListener.failed(new RpcException("Failure while parsing message locally.", e));
+ }
+
+ }
+
+ @Override
+ public void interrupted(InterruptedException e) {
+ innerListener.interrupted(e);
+ }
+
+ }
+ }
+
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/67d5cc6d/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/Controller.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/Controller.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/Controller.java
index af60ff0..94df739 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/Controller.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/Controller.java
@@ -17,10 +17,17 @@
*/
package org.apache.drill.exec.rpc.control;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.DrillBuf;
+
import java.io.Closeable;
import org.apache.drill.exec.exception.DrillbitStartupException;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.rpc.UserRpcException;
+
+import com.google.protobuf.MessageLite;
+import com.google.protobuf.Parser;
/**
* Service that allows one Drillbit to communicate with another. Internally manages whether each particular bit is a
@@ -41,5 +48,61 @@ public interface Controller extends Closeable {
public DrillbitEndpoint start(DrillbitEndpoint partialEndpoint) throws DrillbitStartupException;
+ /**
+ * Register a new handler for custom message types. Should be done before any messages. This is threadsafe as this
+ * method manages locking internally.
+ *
+ * @param messageTypeId
+ * The type of message id to handle. This corresponds to the CustomMessage.type field. Note that only a
+ * single handler for a particular type of message can be registered within a particular Drillbit.
+ * @param handler
+ * The handler that should be used to handle this type of message.
+ * @param parser
+ * The parser used to handle the types of messages the handler above handles.
+ */
+ public <REQUEST extends MessageLite, RESPONSE extends MessageLite> void registerCustomHandler(int messageTypeId,
+ CustomMessageHandler<REQUEST, RESPONSE> handler, Parser<REQUEST> parser);
+
+ /**
+ * Defines how the Controller should handle custom messages. Implementations need to be threadsafe.
+ * @param <REQUEST>
+ * The type of request message.
+ * @param <RESPONSE>
+ * The type of the response message.
+ */
+ public interface CustomMessageHandler<REQUEST extends MessageLite, RESPONSE extends MessageLite> {
+
+ /**
+ * Handle an incoming message.
+ * @param pBody
+ * The protobuf body message object of type REQUEST that was sent.
+ * @param dBody
+ * An optional byte body that was sent along with the structured message.
+ * @return The response that should be sent to the message sender.
+ * @throws UserRpcException
+ * throw this exception if there is an RPC failure that should be communicated to the sender.
+ */
+ public CustomResponse<RESPONSE> onMessage(REQUEST pBody, DrillBuf dBody) throws UserRpcException;
+ }
+
+ /**
+ * A simple interface that describes the nature of the response to the custom incoming message.
+ *
+ * @param <RESPONSE>
+ * The type of message that the respopnse contains. Must be a protobuf message type.
+ */
+ public interface CustomResponse<RESPONSE extends MessageLite> {
+
+ /**
+ * The structured portion of the response.
+ * @return A protobuf message of type RESPONSE
+ */
+ public RESPONSE getMessage();
+ /**
+ * The optional unstructured portion of the message.
+ * @return null or one or more unstructured bodies.
+ */
+ public ByteBuf[] getBodies();
+ }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/67d5cc6d/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControllerImpl.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControllerImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControllerImpl.java
index 631d479..0564cca 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControllerImpl.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControllerImpl.java
@@ -24,6 +24,8 @@ import org.apache.drill.exec.server.BootStrapContext;
import org.apache.drill.exec.work.batch.ControlMessageHandler;
import com.google.common.io.Closeables;
+import com.google.protobuf.MessageLite;
+import com.google.protobuf.Parser;
/**
* Manages communication tunnels between nodes.
@@ -36,6 +38,7 @@ public class ControllerImpl implements Controller {
private final BootStrapContext context;
private final ConnectionManagerRegistry connectionRegistry;
private final boolean allowPortHunting;
+ private final CustomHandlerRegistry handlerRegistry;
public ControllerImpl(BootStrapContext context, ControlMessageHandler handler, boolean allowPortHunting) {
super();
@@ -43,6 +46,7 @@ public class ControllerImpl implements Controller {
this.context = context;
this.connectionRegistry = new ConnectionManagerRegistry(handler, context);
this.allowPortHunting = allowPortHunting;
+ this.handlerRegistry = handler.getHandlerRegistry();
}
@Override
@@ -52,6 +56,7 @@ public class ControllerImpl implements Controller {
port = server.bind(port, allowPortHunting);
DrillbitEndpoint completeEndpoint = partialEndpoint.toBuilder().setControlPort(port).build();
connectionRegistry.setEndpoint(completeEndpoint);
+ handlerRegistry.setEndpoint(completeEndpoint);
return completeEndpoint;
}
@@ -60,6 +65,13 @@ public class ControllerImpl implements Controller {
return new ControlTunnel(endpoint, connectionRegistry.getConnectionManager(endpoint));
}
+
+ @Override
+ public <REQUEST extends MessageLite, RESPONSE extends MessageLite> void registerCustomHandler(int messageTypeId,
+ CustomMessageHandler<REQUEST, RESPONSE> handler, Parser<REQUEST> parser) {
+ handlerRegistry.registerCustomHandler(messageTypeId, handler, parser);
+ }
+
public void close() {
Closeables.closeQuietly(server);
for (ControlConnectionManager bt : connectionRegistry) {
@@ -67,4 +79,5 @@ public class ControllerImpl implements Controller {
}
}
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/67d5cc6d/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/CustomHandlerRegistry.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/CustomHandlerRegistry.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/CustomHandlerRegistry.java
new file mode 100644
index 0000000..c328cd8
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/CustomHandlerRegistry.java
@@ -0,0 +1,124 @@
+/**
+ * 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.drill.exec.rpc.control;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.DrillBuf;
+
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.drill.common.concurrent.AutoCloseableLock;
+import org.apache.drill.exec.proto.BitControl.CustomMessage;
+import org.apache.drill.exec.proto.BitControl.RpcType;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.rpc.Response;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.UserRpcException;
+import org.apache.drill.exec.rpc.control.Controller.CustomMessageHandler;
+import org.apache.drill.exec.rpc.control.Controller.CustomResponse;
+
+import com.carrotsearch.hppc.IntObjectOpenHashMap;
+import com.google.common.base.Preconditions;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.MessageLite;
+import com.google.protobuf.Parser;
+
+public class CustomHandlerRegistry {
+ // private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(CustomHandlerRegistry.class);
+
+ private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
+ private final AutoCloseableLock read = new AutoCloseableLock(readWriteLock.readLock());
+ private final AutoCloseableLock write = new AutoCloseableLock(readWriteLock.writeLock());
+ private final IntObjectOpenHashMap<ParsingHandler<?>> handlers = new IntObjectOpenHashMap<>();
+ private volatile DrillbitEndpoint endpoint;
+
+ public CustomHandlerRegistry() {
+ }
+
+ public void setEndpoint(DrillbitEndpoint endpoint) {
+ this.endpoint = endpoint;
+ }
+
+ public <SEND extends MessageLite> void registerCustomHandler(int messageTypeId,
+ CustomMessageHandler<SEND, ?> handler,
+ Parser<SEND> parser) {
+ Preconditions.checkNotNull(handler);
+ Preconditions.checkNotNull(parser);
+ try (AutoCloseableLock lock = write.open()) {
+ ParsingHandler<?> parsingHandler = handlers.get(messageTypeId);
+ if (parsingHandler != null) {
+ throw new IllegalStateException(String.format(
+ "Only one handler can be registered for a given custom message type. You tried to register a handler for "
+ + "the %d message type but one had already been registered.",
+ messageTypeId));
+ }
+
+ parsingHandler = new ParsingHandler<SEND>(handler, parser);
+ handlers.put(messageTypeId, parsingHandler);
+ }
+ }
+
+ public Response handle(CustomMessage message, DrillBuf dBody) throws RpcException {
+ final ParsingHandler<?> handler;
+ try (AutoCloseableLock lock = read.open()) {
+ handler = handlers.get(message.getType());
+ }
+
+ if (handler == null) {
+ throw new UserRpcException(
+ endpoint, "Unable to handle message.",
+ new IllegalStateException(String.format(
+ "Unable to handle message. The message type provided [%d] did not have a registered handler.",
+ message.getType())));
+ }
+ final CustomResponse<?> customResponse = handler.onMessage(message.getMessage(), dBody);
+ final CustomMessage responseMessage = CustomMessage.newBuilder()
+ .setMessage(customResponse.getMessage().toByteString())
+ .setType(message.getType())
+ .build();
+ // make sure we don't pass in a null array.
+ final ByteBuf[] dBodies = customResponse.getBodies() == null ? new DrillBuf[0] : customResponse.getBodies();
+ return new Response(RpcType.RESP_CUSTOM, responseMessage, dBodies);
+
+ }
+
+ private class ParsingHandler<SEND extends MessageLite> {
+ private final CustomMessageHandler<SEND, ?> handler;
+ private final Parser<SEND> parser;
+
+ public ParsingHandler(CustomMessageHandler<SEND, ?> handler, Parser<SEND> parser) {
+ super();
+ this.handler = handler;
+ this.parser = parser;
+ }
+
+ public CustomResponse<?> onMessage(ByteString pBody, DrillBuf dBody) throws UserRpcException {
+
+ try {
+ final SEND message = parser.parseFrom(pBody);
+ return handler.onMessage(message, dBody);
+
+ } catch (InvalidProtocolBufferException e) {
+ throw new UserRpcException(endpoint, "Failure parsing message.", e);
+ }
+
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/67d5cc6d/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/DefaultInstanceHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/DefaultInstanceHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/DefaultInstanceHandler.java
index 10fe343..7065201 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/DefaultInstanceHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/DefaultInstanceHandler.java
@@ -19,6 +19,7 @@ package org.apache.drill.exec.rpc.control;
import org.apache.drill.exec.proto.BitControl.BitControlHandshake;
import org.apache.drill.exec.proto.BitControl.BitStatus;
+import org.apache.drill.exec.proto.BitControl.CustomMessage;
import org.apache.drill.exec.proto.BitControl.FragmentStatus;
import org.apache.drill.exec.proto.BitControl.RpcType;
import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
@@ -46,7 +47,8 @@ public class DefaultInstanceHandler {
return BitStatus.getDefaultInstance();
case RpcType.RESP_QUERY_STATUS_VALUE:
return QueryProfile.getDefaultInstance();
-
+ case RpcType.RESP_CUSTOM_VALUE:
+ return CustomMessage.getDefaultInstance();
default:
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/67d5cc6d/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/ControlMessageHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/ControlMessageHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/ControlMessageHandler.java
index 1c0eb80..77c069b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/ControlMessageHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/ControlMessageHandler.java
@@ -19,8 +19,10 @@ package org.apache.drill.exec.work.batch;
import static org.apache.drill.exec.rpc.RpcBus.get;
import io.netty.buffer.ByteBuf;
+import io.netty.buffer.DrillBuf;
import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.proto.BitControl.CustomMessage;
import org.apache.drill.exec.proto.BitControl.FinishedReceiver;
import org.apache.drill.exec.proto.BitControl.FragmentStatus;
import org.apache.drill.exec.proto.BitControl.InitializeFragments;
@@ -39,6 +41,7 @@ import org.apache.drill.exec.rpc.UserRpcException;
import org.apache.drill.exec.rpc.control.ControlConnection;
import org.apache.drill.exec.rpc.control.ControlRpcConfig;
import org.apache.drill.exec.rpc.control.ControlTunnel;
+import org.apache.drill.exec.rpc.control.CustomHandlerRegistry;
import org.apache.drill.exec.server.DrillbitContext;
import org.apache.drill.exec.work.WorkManager.WorkerBee;
import org.apache.drill.exec.work.foreman.Foreman;
@@ -50,6 +53,7 @@ import org.apache.drill.exec.work.fragment.NonRootFragmentManager;
public class ControlMessageHandler {
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ControlMessageHandler.class);
private final WorkerBee bee;
+ private final CustomHandlerRegistry handlerRegistry = new CustomHandlerRegistry();
public ControlMessageHandler(final WorkerBee bee) {
this.bee = bee;
@@ -69,6 +73,11 @@ public class ControlMessageHandler {
return ControlRpcConfig.OK;
}
+ case RpcType.REQ_CUSTOM_VALUE: {
+ final CustomMessage customMessage = get(pBody, CustomMessage.PARSER);
+ return handlerRegistry.handle(customMessage, (DrillBuf) dBody);
+ }
+
case RpcType.REQ_RECEIVER_FINISHED_VALUE: {
final FinishedReceiver finishedReceiver = get(pBody, FinishedReceiver.PARSER);
receivingFragmentFinished(finishedReceiver);
@@ -228,4 +237,9 @@ public class ControlMessageHandler {
return Acks.OK;
}
+
+ public CustomHandlerRegistry getHandlerRegistry() {
+ return handlerRegistry;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/67d5cc6d/exec/java-exec/src/test/java/org/apache/drill/exec/rpc/control/TestCustomTunnel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/rpc/control/TestCustomTunnel.java b/exec/java-exec/src/test/java/org/apache/drill/exec/rpc/control/TestCustomTunnel.java
new file mode 100644
index 0000000..2008a48
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/rpc/control/TestCustomTunnel.java
@@ -0,0 +1,138 @@
+/**
+ * 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.drill.exec.rpc.control;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.DrillBuf;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import io.netty.util.internal.ThreadLocalRandom;
+
+import java.util.Arrays;
+import java.util.Random;
+
+import org.apache.drill.BaseTestQuery;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.UserRpcException;
+import org.apache.drill.exec.rpc.control.ControlTunnel.CustomFuture;
+import org.apache.drill.exec.rpc.control.ControlTunnel.CustomTunnel;
+import org.apache.drill.exec.rpc.control.Controller.CustomMessageHandler;
+import org.apache.drill.exec.rpc.control.Controller.CustomResponse;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.junit.Test;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+
+public class TestCustomTunnel extends BaseTestQuery {
+
+ private final QueryId expectedId = QueryId
+ .newBuilder()
+ .setPart1(ThreadLocalRandom.current().nextLong())
+ .setPart2(ThreadLocalRandom.current().nextLong())
+ .build();
+
+ private final ByteBuf buf1;
+ private final byte[] expected;
+
+ public TestCustomTunnel() {
+ buf1 = UnpooledByteBufAllocator.DEFAULT.buffer(1024);
+ Random r = new Random();
+ this.expected = new byte[1024];
+ r.nextBytes(expected);
+ buf1.writeBytes(expected);
+ }
+
+ @Test
+ public void ensureRoundTrip() throws RpcException, InvalidProtocolBufferException {
+
+ final DrillbitContext context = getDrillbitContext();
+ final TestCustomMessageHandler handler = new TestCustomMessageHandler(context.getEndpoint(), false);
+ context.getController().registerCustomHandler(1001, handler, DrillbitEndpoint.PARSER);
+ final ControlTunnel loopbackTunnel = context.getController().getTunnel(context.getEndpoint());
+ final CustomTunnel<DrillbitEndpoint, QueryId> tunnel = loopbackTunnel.getCustomTunnel(1001, DrillbitEndpoint.class,
+ QueryId.PARSER);
+ CustomFuture<QueryId> future = tunnel.send(context.getEndpoint());
+ assertEquals(expectedId, future.get());
+ }
+
+ @Test
+ public void ensureRoundTripBytes() throws RpcException, InvalidProtocolBufferException {
+ final DrillbitContext context = getDrillbitContext();
+ final TestCustomMessageHandler handler = new TestCustomMessageHandler(context.getEndpoint(), true);
+ context.getController().registerCustomHandler(1002, handler, DrillbitEndpoint.PARSER);
+ final ControlTunnel loopbackTunnel = context.getController().getTunnel(context.getEndpoint());
+ final CustomTunnel<DrillbitEndpoint, QueryId> tunnel = loopbackTunnel.getCustomTunnel(1002, DrillbitEndpoint.class,
+ QueryId.PARSER);
+ buf1.retain();
+ CustomFuture<QueryId> future = tunnel.send(context.getEndpoint(), buf1);
+ assertEquals(expectedId, future.get());
+ byte[] actual = new byte[1024];
+ future.getBuffer().getBytes(0, actual);
+ future.getBuffer().release();
+ assertTrue(Arrays.equals(expected, actual));
+ }
+
+ private class TestCustomMessageHandler implements CustomMessageHandler<DrillbitEndpoint, QueryId> {
+ private DrillbitEndpoint expectedValue;
+ private final boolean returnBytes;
+
+ public TestCustomMessageHandler(DrillbitEndpoint expectedValue, boolean returnBytes) {
+ super();
+ this.expectedValue = expectedValue;
+ this.returnBytes = returnBytes;
+ }
+
+ @Override
+ public CustomResponse<QueryId> onMessage(DrillbitEndpoint pBody, DrillBuf dBody) throws UserRpcException {
+
+ if (!expectedValue.equals(pBody)) {
+ throw new UserRpcException(expectedValue, "Invalid expected downstream value.", new IllegalStateException());
+ }
+
+ if (returnBytes) {
+ byte[] actual = new byte[1024];
+ dBody.getBytes(0, actual);
+ if (!Arrays.equals(expected, actual)) {
+ throw new UserRpcException(expectedValue, "Invalid expected downstream value.", new IllegalStateException());
+ }
+ }
+
+ return new CustomResponse<QueryId>() {
+
+ @Override
+ public QueryId getMessage() {
+ return expectedId;
+ }
+
+ @Override
+ public ByteBuf[] getBodies() {
+ if (returnBytes) {
+ buf1.retain();
+ return new ByteBuf[] { buf1 };
+ } else {
+ return null;
+ }
+ }
+
+ };
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/67d5cc6d/protocol/src/main/java/org/apache/drill/exec/proto/BitControl.java
----------------------------------------------------------------------
diff --git a/protocol/src/main/java/org/apache/drill/exec/proto/BitControl.java b/protocol/src/main/java/org/apache/drill/exec/proto/BitControl.java
index 493036b..b16934d 100644
--- a/protocol/src/main/java/org/apache/drill/exec/proto/BitControl.java
+++ b/protocol/src/main/java/org/apache/drill/exec/proto/BitControl.java
@@ -99,25 +99,33 @@ public final class BitControl {
*/
REQ_UNPAUSE_FRAGMENT(10, 16),
/**
+ * <code>REQ_CUSTOM = 17;</code>
+ */
+ REQ_CUSTOM(11, 17),
+ /**
* <code>RESP_FRAGMENT_HANDLE = 11;</code>
*
* <pre>
* bit responses
* </pre>
*/
- RESP_FRAGMENT_HANDLE(11, 11),
+ RESP_FRAGMENT_HANDLE(12, 11),
/**
* <code>RESP_FRAGMENT_STATUS = 12;</code>
*/
- RESP_FRAGMENT_STATUS(12, 12),
+ RESP_FRAGMENT_STATUS(13, 12),
/**
* <code>RESP_BIT_STATUS = 13;</code>
*/
- RESP_BIT_STATUS(13, 13),
+ RESP_BIT_STATUS(14, 13),
/**
* <code>RESP_QUERY_STATUS = 14;</code>
*/
- RESP_QUERY_STATUS(14, 14),
+ RESP_QUERY_STATUS(15, 14),
+ /**
+ * <code>RESP_CUSTOM = 18;</code>
+ */
+ RESP_CUSTOM(16, 18),
;
/**
@@ -185,6 +193,10 @@ public final class BitControl {
*/
public static final int REQ_UNPAUSE_FRAGMENT_VALUE = 16;
/**
+ * <code>REQ_CUSTOM = 17;</code>
+ */
+ public static final int REQ_CUSTOM_VALUE = 17;
+ /**
* <code>RESP_FRAGMENT_HANDLE = 11;</code>
*
* <pre>
@@ -204,6 +216,10 @@ public final class BitControl {
* <code>RESP_QUERY_STATUS = 14;</code>
*/
public static final int RESP_QUERY_STATUS_VALUE = 14;
+ /**
+ * <code>RESP_CUSTOM = 18;</code>
+ */
+ public static final int RESP_CUSTOM_VALUE = 18;
public final int getNumber() { return value; }
@@ -221,10 +237,12 @@ public final class BitControl {
case 10: return REQ_QUERY_STATUS;
case 15: return REQ_QUERY_CANCEL;
case 16: return REQ_UNPAUSE_FRAGMENT;
+ case 17: return REQ_CUSTOM;
case 11: return RESP_FRAGMENT_HANDLE;
case 12: return RESP_FRAGMENT_STATUS;
case 13: return RESP_BIT_STATUS;
case 14: return RESP_QUERY_STATUS;
+ case 18: return RESP_CUSTOM;
default: return null;
}
}
@@ -3001,6 +3019,486 @@ public final class BitControl {
// @@protoc_insertion_point(class_scope:exec.bit.control.InitializeFragments)
}
+ public interface CustomMessageOrBuilder
+ extends com.google.protobuf.MessageOrBuilder {
+
+ // optional int32 type = 1;
+ /**
+ * <code>optional int32 type = 1;</code>
+ */
+ boolean hasType();
+ /**
+ * <code>optional int32 type = 1;</code>
+ */
+ int getType();
+
+ // optional bytes message = 2;
+ /**
+ * <code>optional bytes message = 2;</code>
+ */
+ boolean hasMessage();
+ /**
+ * <code>optional bytes message = 2;</code>
+ */
+ com.google.protobuf.ByteString getMessage();
+ }
+ /**
+ * Protobuf type {@code exec.bit.control.CustomMessage}
+ */
+ public static final class CustomMessage extends
+ com.google.protobuf.GeneratedMessage
+ implements CustomMessageOrBuilder {
+ // Use CustomMessage.newBuilder() to construct.
+ private CustomMessage(com.google.protobuf.GeneratedMessage.Builder<?> builder) {
+ super(builder);
+ this.unknownFields = builder.getUnknownFields();
+ }
+ private CustomMessage(boolean noInit) { this.unknownFields = com.google.protobuf.UnknownFieldSet.getDefaultInstance(); }
+
+ private static final CustomMessage defaultInstance;
+ public static CustomMessage getDefaultInstance() {
+ return defaultInstance;
+ }
+
+ public CustomMessage getDefaultInstanceForType() {
+ return defaultInstance;
+ }
+
+ private final com.google.protobuf.UnknownFieldSet unknownFields;
+ @java.lang.Override
+ public final com.google.protobuf.UnknownFieldSet
+ getUnknownFields() {
+ return this.unknownFields;
+ }
+ private CustomMessage(
+ com.google.protobuf.CodedInputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ initFields();
+ int mutable_bitField0_ = 0;
+ com.google.protobuf.UnknownFieldSet.Builder unknownFields =
+ com.google.protobuf.UnknownFieldSet.newBuilder();
+ try {
+ boolean done = false;
+ while (!done) {
+ int tag = input.readTag();
+ switch (tag) {
+ case 0:
+ done = true;
+ break;
+ default: {
+ if (!parseUnknownField(input, unknownFields,
+ extensionRegistry, tag)) {
+ done = true;
+ }
+ break;
+ }
+ case 8: {
+ bitField0_ |= 0x00000001;
+ type_ = input.readInt32();
+ break;
+ }
+ case 18: {
+ bitField0_ |= 0x00000002;
+ message_ = input.readBytes();
+ break;
+ }
+ }
+ }
+ } catch (com.google.protobuf.InvalidProtocolBufferException e) {
+ throw e.setUnfinishedMessage(this);
+ } catch (java.io.IOException e) {
+ throw new com.google.protobuf.InvalidProtocolBufferException(
+ e.getMessage()).setUnfinishedMessage(this);
+ } finally {
+ this.unknownFields = unknownFields.build();
+ makeExtensionsImmutable();
+ }
+ }
+ public static final com.google.protobuf.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.drill.exec.proto.BitControl.internal_static_exec_bit_control_CustomMessage_descriptor;
+ }
+
+ protected com.google.protobuf.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.drill.exec.proto.BitControl.internal_static_exec_bit_control_CustomMessage_fieldAccessorTable
+ .ensureFieldAccessorsInitialized(
+ org.apache.drill.exec.proto.BitControl.CustomMessage.class, org.apache.drill.exec.proto.BitControl.CustomMessage.Builder.class);
+ }
+
+ public static com.google.protobuf.Parser<CustomMessage> PARSER =
+ new com.google.protobuf.AbstractParser<CustomMessage>() {
+ public CustomMessage parsePartialFrom(
+ com.google.protobuf.CodedInputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ return new CustomMessage(input, extensionRegistry);
+ }
+ };
+
+ @java.lang.Override
+ public com.google.protobuf.Parser<CustomMessage> getParserForType() {
+ return PARSER;
+ }
+
+ private int bitField0_;
+ // optional int32 type = 1;
+ public static final int TYPE_FIELD_NUMBER = 1;
+ private int type_;
+ /**
+ * <code>optional int32 type = 1;</code>
+ */
+ public boolean hasType() {
+ return ((bitField0_ & 0x00000001) == 0x00000001);
+ }
+ /**
+ * <code>optional int32 type = 1;</code>
+ */
+ public int getType() {
+ return type_;
+ }
+
+ // optional bytes message = 2;
+ public static final int MESSAGE_FIELD_NUMBER = 2;
+ private com.google.protobuf.ByteString message_;
+ /**
+ * <code>optional bytes message = 2;</code>
+ */
+ public boolean hasMessage() {
+ return ((bitField0_ & 0x00000002) == 0x00000002);
+ }
+ /**
+ * <code>optional bytes message = 2;</code>
+ */
+ public com.google.protobuf.ByteString getMessage() {
+ return message_;
+ }
+
+ private void initFields() {
+ type_ = 0;
+ message_ = com.google.protobuf.ByteString.EMPTY;
+ }
+ private byte memoizedIsInitialized = -1;
+ public final boolean isInitialized() {
+ byte isInitialized = memoizedIsInitialized;
+ if (isInitialized != -1) return isInitialized == 1;
+
+ memoizedIsInitialized = 1;
+ return true;
+ }
+
+ public void writeTo(com.google.protobuf.CodedOutputStream output)
+ throws java.io.IOException {
+ getSerializedSize();
+ if (((bitField0_ & 0x00000001) == 0x00000001)) {
+ output.writeInt32(1, type_);
+ }
+ if (((bitField0_ & 0x00000002) == 0x00000002)) {
+ output.writeBytes(2, message_);
+ }
+ getUnknownFields().writeTo(output);
+ }
+
+ private int memoizedSerializedSize = -1;
+ public int getSerializedSize() {
+ int size = memoizedSerializedSize;
+ if (size != -1) return size;
+
+ size = 0;
+ if (((bitField0_ & 0x00000001) == 0x00000001)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeInt32Size(1, type_);
+ }
+ if (((bitField0_ & 0x00000002) == 0x00000002)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeBytesSize(2, message_);
+ }
+ size += getUnknownFields().getSerializedSize();
+ memoizedSerializedSize = size;
+ return size;
+ }
+
+ private static final long serialVersionUID = 0L;
+ @java.lang.Override
+ protected java.lang.Object writeReplace()
+ throws java.io.ObjectStreamException {
+ return super.writeReplace();
+ }
+
+ public static org.apache.drill.exec.proto.BitControl.CustomMessage parseFrom(
+ com.google.protobuf.ByteString data)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.drill.exec.proto.BitControl.CustomMessage parseFrom(
+ com.google.protobuf.ByteString data,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.drill.exec.proto.BitControl.CustomMessage parseFrom(byte[] data)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data);
+ }
+ public static org.apache.drill.exec.proto.BitControl.CustomMessage parseFrom(
+ byte[] data,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws com.google.protobuf.InvalidProtocolBufferException {
+ return PARSER.parseFrom(data, extensionRegistry);
+ }
+ public static org.apache.drill.exec.proto.BitControl.CustomMessage parseFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ return PARSER.parseFrom(input);
+ }
+ public static org.apache.drill.exec.proto.BitControl.CustomMessage parseFrom(
+ java.io.InputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return PARSER.parseFrom(input, extensionRegistry);
+ }
+ public static org.apache.drill.exec.proto.BitControl.CustomMessage parseDelimitedFrom(java.io.InputStream input)
+ throws java.io.IOException {
+ return PARSER.parseDelimitedFrom(input);
+ }
+ public static org.apache.drill.exec.proto.BitControl.CustomMessage parseDelimitedFrom(
+ java.io.InputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return PARSER.parseDelimitedFrom(input, extensionRegistry);
+ }
+ public static org.apache.drill.exec.proto.BitControl.CustomMessage parseFrom(
+ com.google.protobuf.CodedInputStream input)
+ throws java.io.IOException {
+ return PARSER.parseFrom(input);
+ }
+ public static org.apache.drill.exec.proto.BitControl.CustomMessage parseFrom(
+ com.google.protobuf.CodedInputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ return PARSER.parseFrom(input, extensionRegistry);
+ }
+
+ public static Builder newBuilder() { return Builder.create(); }
+ public Builder newBuilderForType() { return newBuilder(); }
+ public static Builder newBuilder(org.apache.drill.exec.proto.BitControl.CustomMessage prototype) {
+ return newBuilder().mergeFrom(prototype);
+ }
+ public Builder toBuilder() { return newBuilder(this); }
+
+ @java.lang.Override
+ protected Builder newBuilderForType(
+ com.google.protobuf.GeneratedMessage.BuilderParent parent) {
+ Builder builder = new Builder(parent);
+ return builder;
+ }
+ /**
+ * Protobuf type {@code exec.bit.control.CustomMessage}
+ */
+ public static final class Builder extends
+ com.google.protobuf.GeneratedMessage.Builder<Builder>
+ implements org.apache.drill.exec.proto.BitControl.CustomMessageOrBuilder {
+ public static final com.google.protobuf.Descriptors.Descriptor
+ getDescriptor() {
+ return org.apache.drill.exec.proto.BitControl.internal_static_exec_bit_control_CustomMessage_descriptor;
+ }
+
+ protected com.google.protobuf.GeneratedMessage.FieldAccessorTable
+ internalGetFieldAccessorTable() {
+ return org.apache.drill.exec.proto.BitControl.internal_static_exec_bit_control_CustomMessage_fieldAccessorTable
+ .ensureFieldAccessorsInitialized(
+ org.apache.drill.exec.proto.BitControl.CustomMessage.class, org.apache.drill.exec.proto.BitControl.CustomMessage.Builder.class);
+ }
+
+ // Construct using org.apache.drill.exec.proto.BitControl.CustomMessage.newBuilder()
+ private Builder() {
+ maybeForceBuilderInitialization();
+ }
+
+ private Builder(
+ com.google.protobuf.GeneratedMessage.BuilderParent parent) {
+ super(parent);
+ maybeForceBuilderInitialization();
+ }
+ private void maybeForceBuilderInitialization() {
+ if (com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders) {
+ }
+ }
+ private static Builder create() {
+ return new Builder();
+ }
+
+ public Builder clear() {
+ super.clear();
+ type_ = 0;
+ bitField0_ = (bitField0_ & ~0x00000001);
+ message_ = com.google.protobuf.ByteString.EMPTY;
+ bitField0_ = (bitField0_ & ~0x00000002);
+ return this;
+ }
+
+ public Builder clone() {
+ return create().mergeFrom(buildPartial());
+ }
+
+ public com.google.protobuf.Descriptors.Descriptor
+ getDescriptorForType() {
+ return org.apache.drill.exec.proto.BitControl.internal_static_exec_bit_control_CustomMessage_descriptor;
+ }
+
+ public org.apache.drill.exec.proto.BitControl.CustomMessage getDefaultInstanceForType() {
+ return org.apache.drill.exec.proto.BitControl.CustomMessage.getDefaultInstance();
+ }
+
+ public org.apache.drill.exec.proto.BitControl.CustomMessage build() {
+ org.apache.drill.exec.proto.BitControl.CustomMessage result = buildPartial();
+ if (!result.isInitialized()) {
+ throw newUninitializedMessageException(result);
+ }
+ return result;
+ }
+
+ public org.apache.drill.exec.proto.BitControl.CustomMessage buildPartial() {
+ org.apache.drill.exec.proto.BitControl.CustomMessage result = new org.apache.drill.exec.proto.BitControl.CustomMessage(this);
+ int from_bitField0_ = bitField0_;
+ int to_bitField0_ = 0;
+ if (((from_bitField0_ & 0x00000001) == 0x00000001)) {
+ to_bitField0_ |= 0x00000001;
+ }
+ result.type_ = type_;
+ if (((from_bitField0_ & 0x00000002) == 0x00000002)) {
+ to_bitField0_ |= 0x00000002;
+ }
+ result.message_ = message_;
+ result.bitField0_ = to_bitField0_;
+ onBuilt();
+ return result;
+ }
+
+ public Builder mergeFrom(com.google.protobuf.Message other) {
+ if (other instanceof org.apache.drill.exec.proto.BitControl.CustomMessage) {
+ return mergeFrom((org.apache.drill.exec.proto.BitControl.CustomMessage)other);
+ } else {
+ super.mergeFrom(other);
+ return this;
+ }
+ }
+
+ public Builder mergeFrom(org.apache.drill.exec.proto.BitControl.CustomMessage other) {
+ if (other == org.apache.drill.exec.proto.BitControl.CustomMessage.getDefaultInstance()) return this;
+ if (other.hasType()) {
+ setType(other.getType());
+ }
+ if (other.hasMessage()) {
+ setMessage(other.getMessage());
+ }
+ this.mergeUnknownFields(other.getUnknownFields());
+ return this;
+ }
+
+ public final boolean isInitialized() {
+ return true;
+ }
+
+ public Builder mergeFrom(
+ com.google.protobuf.CodedInputStream input,
+ com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+ throws java.io.IOException {
+ org.apache.drill.exec.proto.BitControl.CustomMessage parsedMessage = null;
+ try {
+ parsedMessage = PARSER.parsePartialFrom(input, extensionRegistry);
+ } catch (com.google.protobuf.InvalidProtocolBufferException e) {
+ parsedMessage = (org.apache.drill.exec.proto.BitControl.CustomMessage) e.getUnfinishedMessage();
+ throw e;
+ } finally {
+ if (parsedMessage != null) {
+ mergeFrom(parsedMessage);
+ }
+ }
+ return this;
+ }
+ private int bitField0_;
+
+ // optional int32 type = 1;
+ private int type_ ;
+ /**
+ * <code>optional int32 type = 1;</code>
+ */
+ public boolean hasType() {
+ return ((bitField0_ & 0x00000001) == 0x00000001);
+ }
+ /**
+ * <code>optional int32 type = 1;</code>
+ */
+ public int getType() {
+ return type_;
+ }
+ /**
+ * <code>optional int32 type = 1;</code>
+ */
+ public Builder setType(int value) {
+ bitField0_ |= 0x00000001;
+ type_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>optional int32 type = 1;</code>
+ */
+ public Builder clearType() {
+ bitField0_ = (bitField0_ & ~0x00000001);
+ type_ = 0;
+ onChanged();
+ return this;
+ }
+
+ // optional bytes message = 2;
+ private com.google.protobuf.ByteString message_ = com.google.protobuf.ByteString.EMPTY;
+ /**
+ * <code>optional bytes message = 2;</code>
+ */
+ public boolean hasMessage() {
+ return ((bitField0_ & 0x00000002) == 0x00000002);
+ }
+ /**
+ * <code>optional bytes message = 2;</code>
+ */
+ public com.google.protobuf.ByteString getMessage() {
+ return message_;
+ }
+ /**
+ * <code>optional bytes message = 2;</code>
+ */
+ public Builder setMessage(com.google.protobuf.ByteString value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ bitField0_ |= 0x00000002;
+ message_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>optional bytes message = 2;</code>
+ */
+ public Builder clearMessage() {
+ bitField0_ = (bitField0_ & ~0x00000002);
+ message_ = getDefaultInstance().getMessage();
+ onChanged();
+ return this;
+ }
+
+ // @@protoc_insertion_point(builder_scope:exec.bit.control.CustomMessage)
+ }
+
+ static {
+ defaultInstance = new CustomMessage(true);
+ defaultInstance.initFields();
+ }
+
+ // @@protoc_insertion_point(class_scope:exec.bit.control.CustomMessage)
+ }
+
public interface PlanFragmentOrBuilder
extends com.google.protobuf.MessageOrBuilder {
@@ -8440,6 +8938,11 @@ public final class BitControl {
com.google.protobuf.GeneratedMessage.FieldAccessorTable
internal_static_exec_bit_control_InitializeFragments_fieldAccessorTable;
private static com.google.protobuf.Descriptors.Descriptor
+ internal_static_exec_bit_control_CustomMessage_descriptor;
+ private static
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable
+ internal_static_exec_bit_control_CustomMessage_fieldAccessorTable;
+ private static com.google.protobuf.Descriptors.Descriptor
internal_static_exec_bit_control_PlanFragment_descriptor;
private static
com.google.protobuf.GeneratedMessage.FieldAccessorTable
@@ -8485,41 +8988,43 @@ public final class BitControl {
"red.MinorFragmentProfile\022(\n\006handle\030\002 \001(\013",
"2\030.exec.bit.FragmentHandle\"G\n\023Initialize" +
"Fragments\0220\n\010fragment\030\001 \003(\0132\036.exec.bit.c" +
- "ontrol.PlanFragment\"\374\003\n\014PlanFragment\022(\n\006" +
- "handle\030\001 \001(\0132\030.exec.bit.FragmentHandle\022\024" +
- "\n\014network_cost\030\004 \001(\002\022\020\n\010cpu_cost\030\005 \001(\002\022\021" +
- "\n\tdisk_cost\030\006 \001(\002\022\023\n\013memory_cost\030\007 \001(\002\022\025" +
- "\n\rfragment_json\030\010 \001(\t\022\025\n\rleaf_fragment\030\t" +
- " \001(\010\022*\n\nassignment\030\n \001(\0132\026.exec.Drillbit" +
- "Endpoint\022\'\n\007foreman\030\013 \001(\0132\026.exec.Drillbi" +
- "tEndpoint\022\035\n\013mem_initial\030\014 \001(\003:\01020000000",
- "\022\033\n\007mem_max\030\r \001(\003:\n2000000000\0221\n\013credent" +
- "ials\030\016 \001(\0132\034.exec.shared.UserCredentials" +
- "\022\024\n\014options_json\030\017 \001(\t\022:\n\007context\030\020 \001(\0132" +
- ").exec.bit.control.QueryContextInformati" +
- "on\022.\n\tcollector\030\021 \003(\0132\033.exec.bit.control" +
- ".Collector\"\210\001\n\tCollector\022\"\n\032opposite_maj" +
- "or_fragment_id\030\001 \001(\005\022#\n\027incoming_minor_f" +
- "ragment\030\002 \003(\005B\002\020\001\022\035\n\025supports_out_of_ord" +
- "er\030\003 \001(\010\022\023\n\013is_spooling\030\004 \001(\010\"c\n\027QueryCo" +
- "ntextInformation\022\030\n\020query_start_time\030\001 \001",
- "(\003\022\021\n\ttime_zone\030\002 \001(\005\022\033\n\023default_schema_" +
- "name\030\003 \001(\t\"f\n\017WorkQueueStatus\022(\n\010endpoin" +
- "t\030\001 \001(\0132\026.exec.DrillbitEndpoint\022\024\n\014queue" +
- "_length\030\002 \001(\005\022\023\n\013report_time\030\003 \001(\003\"h\n\020Fi" +
- "nishedReceiver\022*\n\010receiver\030\001 \001(\0132\030.exec." +
- "bit.FragmentHandle\022(\n\006sender\030\002 \001(\0132\030.exe" +
- "c.bit.FragmentHandle*\323\002\n\007RpcType\022\r\n\tHAND" +
- "SHAKE\020\000\022\007\n\003ACK\020\001\022\013\n\007GOODBYE\020\002\022\034\n\030REQ_INI" +
- "TIALIZE_FRAGMENTS\020\003\022\027\n\023REQ_CANCEL_FRAGME" +
- "NT\020\006\022\031\n\025REQ_RECEIVER_FINISHED\020\007\022\027\n\023REQ_F",
- "RAGMENT_STATUS\020\010\022\022\n\016REQ_BIT_STATUS\020\t\022\024\n\020" +
- "REQ_QUERY_STATUS\020\n\022\024\n\020REQ_QUERY_CANCEL\020\017" +
- "\022\030\n\024REQ_UNPAUSE_FRAGMENT\020\020\022\030\n\024RESP_FRAGM" +
- "ENT_HANDLE\020\013\022\030\n\024RESP_FRAGMENT_STATUS\020\014\022\023" +
- "\n\017RESP_BIT_STATUS\020\r\022\025\n\021RESP_QUERY_STATUS" +
- "\020\016B+\n\033org.apache.drill.exec.protoB\nBitCo" +
- "ntrolH\001"
+ "ontrol.PlanFragment\".\n\rCustomMessage\022\014\n\004" +
+ "type\030\001 \001(\005\022\017\n\007message\030\002 \001(\014\"\374\003\n\014PlanFrag" +
+ "ment\022(\n\006handle\030\001 \001(\0132\030.exec.bit.Fragment" +
+ "Handle\022\024\n\014network_cost\030\004 \001(\002\022\020\n\010cpu_cost" +
+ "\030\005 \001(\002\022\021\n\tdisk_cost\030\006 \001(\002\022\023\n\013memory_cost" +
+ "\030\007 \001(\002\022\025\n\rfragment_json\030\010 \001(\t\022\025\n\rleaf_fr" +
+ "agment\030\t \001(\010\022*\n\nassignment\030\n \001(\0132\026.exec." +
+ "DrillbitEndpoint\022\'\n\007foreman\030\013 \001(\0132\026.exec",
+ ".DrillbitEndpoint\022\035\n\013mem_initial\030\014 \001(\003:\010" +
+ "20000000\022\033\n\007mem_max\030\r \001(\003:\n2000000000\0221\n" +
+ "\013credentials\030\016 \001(\0132\034.exec.shared.UserCre" +
+ "dentials\022\024\n\014options_json\030\017 \001(\t\022:\n\007contex" +
+ "t\030\020 \001(\0132).exec.bit.control.QueryContextI" +
+ "nformation\022.\n\tcollector\030\021 \003(\0132\033.exec.bit" +
+ ".control.Collector\"\210\001\n\tCollector\022\"\n\032oppo" +
+ "site_major_fragment_id\030\001 \001(\005\022#\n\027incoming" +
+ "_minor_fragment\030\002 \003(\005B\002\020\001\022\035\n\025supports_ou" +
+ "t_of_order\030\003 \001(\010\022\023\n\013is_spooling\030\004 \001(\010\"c\n",
+ "\027QueryContextInformation\022\030\n\020query_start_" +
+ "time\030\001 \001(\003\022\021\n\ttime_zone\030\002 \001(\005\022\033\n\023default" +
+ "_schema_name\030\003 \001(\t\"f\n\017WorkQueueStatus\022(\n" +
+ "\010endpoint\030\001 \001(\0132\026.exec.DrillbitEndpoint\022" +
+ "\024\n\014queue_length\030\002 \001(\005\022\023\n\013report_time\030\003 \001" +
+ "(\003\"h\n\020FinishedReceiver\022*\n\010receiver\030\001 \001(\013" +
+ "2\030.exec.bit.FragmentHandle\022(\n\006sender\030\002 \001" +
+ "(\0132\030.exec.bit.FragmentHandle*\364\002\n\007RpcType" +
+ "\022\r\n\tHANDSHAKE\020\000\022\007\n\003ACK\020\001\022\013\n\007GOODBYE\020\002\022\034\n" +
+ "\030REQ_INITIALIZE_FRAGMENTS\020\003\022\027\n\023REQ_CANCE",
+ "L_FRAGMENT\020\006\022\031\n\025REQ_RECEIVER_FINISHED\020\007\022" +
+ "\027\n\023REQ_FRAGMENT_STATUS\020\010\022\022\n\016REQ_BIT_STAT" +
+ "US\020\t\022\024\n\020REQ_QUERY_STATUS\020\n\022\024\n\020REQ_QUERY_" +
+ "CANCEL\020\017\022\030\n\024REQ_UNPAUSE_FRAGMENT\020\020\022\016\n\nRE" +
+ "Q_CUSTOM\020\021\022\030\n\024RESP_FRAGMENT_HANDLE\020\013\022\030\n\024" +
+ "RESP_FRAGMENT_STATUS\020\014\022\023\n\017RESP_BIT_STATU" +
+ "S\020\r\022\025\n\021RESP_QUERY_STATUS\020\016\022\017\n\013RESP_CUSTO" +
+ "M\020\022B+\n\033org.apache.drill.exec.protoB\nBitC" +
+ "ontrolH\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -8550,32 +9055,38 @@ public final class BitControl {
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_exec_bit_control_InitializeFragments_descriptor,
new java.lang.String[] { "Fragment", });
- internal_static_exec_bit_control_PlanFragment_descriptor =
+ internal_static_exec_bit_control_CustomMessage_descriptor =
getDescriptor().getMessageTypes().get(4);
+ internal_static_exec_bit_control_CustomMessage_fieldAccessorTable = new
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable(
+ internal_static_exec_bit_control_CustomMessage_descriptor,
+ new java.lang.String[] { "Type", "Message", });
+ internal_static_exec_bit_control_PlanFragment_descriptor =
+ getDescriptor().getMessageTypes().get(5);
internal_static_exec_bit_control_PlanFragment_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_exec_bit_control_PlanFragment_descriptor,
new java.lang.String[] { "Handle", "NetworkCost", "CpuCost", "DiskCost", "MemoryCost", "FragmentJson", "LeafFragment", "Assignment", "Foreman", "MemInitial", "MemMax", "Credentials", "OptionsJson", "Context", "Collector", });
internal_static_exec_bit_control_Collector_descriptor =
- getDescriptor().getMessageTypes().get(5);
+ getDescriptor().getMessageTypes().get(6);
internal_static_exec_bit_control_Collector_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_exec_bit_control_Collector_descriptor,
new java.lang.String[] { "OppositeMajorFragmentId", "IncomingMinorFragment", "SupportsOutOfOrder", "IsSpooling", });
internal_static_exec_bit_control_QueryContextInformation_descriptor =
- getDescriptor().getMessageTypes().get(6);
+ getDescriptor().getMessageTypes().get(7);
internal_static_exec_bit_control_QueryContextInformation_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_exec_bit_control_QueryContextInformation_descriptor,
new java.lang.String[] { "QueryStartTime", "TimeZone", "DefaultSchemaName", });
internal_static_exec_bit_control_WorkQueueStatus_descriptor =
- getDescriptor().getMessageTypes().get(7);
+ getDescriptor().getMessageTypes().get(8);
internal_static_exec_bit_control_WorkQueueStatus_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_exec_bit_control_WorkQueueStatus_descriptor,
new java.lang.String[] { "Endpoint", "QueueLength", "ReportTime", });
internal_static_exec_bit_control_FinishedReceiver_descriptor =
- getDescriptor().getMessageTypes().get(8);
+ getDescriptor().getMessageTypes().get(9);
internal_static_exec_bit_control_FinishedReceiver_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_exec_bit_control_FinishedReceiver_descriptor,
http://git-wip-us.apache.org/repos/asf/drill/blob/67d5cc6d/protocol/src/main/java/org/apache/drill/exec/proto/SchemaBitControl.java
----------------------------------------------------------------------
diff --git a/protocol/src/main/java/org/apache/drill/exec/proto/SchemaBitControl.java b/protocol/src/main/java/org/apache/drill/exec/proto/SchemaBitControl.java
index a4088c9..0a9b90d 100644
--- a/protocol/src/main/java/org/apache/drill/exec/proto/SchemaBitControl.java
+++ b/protocol/src/main/java/org/apache/drill/exec/proto/SchemaBitControl.java
@@ -499,6 +499,125 @@ public final class SchemaBitControl
}
}
+ public static final class CustomMessage
+ {
+ public static final org.apache.drill.exec.proto.SchemaBitControl.CustomMessage.MessageSchema WRITE =
+ new org.apache.drill.exec.proto.SchemaBitControl.CustomMessage.MessageSchema();
+ public static final org.apache.drill.exec.proto.SchemaBitControl.CustomMessage.BuilderSchema MERGE =
+ new org.apache.drill.exec.proto.SchemaBitControl.CustomMessage.BuilderSchema();
+
+ public static class MessageSchema implements com.dyuproject.protostuff.Schema<org.apache.drill.exec.proto.BitControl.CustomMessage>
+ {
+ public void writeTo(com.dyuproject.protostuff.Output output, org.apache.drill.exec.proto.BitControl.CustomMessage message) throws java.io.IOException
+ {
+ if(message.hasType())
+ output.writeInt32(1, message.getType(), false);
+ if(message.hasMessage())
+ output.writeByteArray(2, message.getMessage().toByteArray(), false);
+
+ }
+ public boolean isInitialized(org.apache.drill.exec.proto.BitControl.CustomMessage message)
+ {
+ return message.isInitialized();
+ }
+ public java.lang.String getFieldName(int number)
+ {
+ return org.apache.drill.exec.proto.SchemaBitControl.CustomMessage.getFieldName(number);
+ }
+ public int getFieldNumber(java.lang.String name)
+ {
+ return org.apache.drill.exec.proto.SchemaBitControl.CustomMessage.getFieldNumber(name);
+ }
+ public java.lang.Class<org.apache.drill.exec.proto.BitControl.CustomMessage> typeClass()
+ {
+ return org.apache.drill.exec.proto.BitControl.CustomMessage.class;
+ }
+ public java.lang.String messageName()
+ {
+ return org.apache.drill.exec.proto.BitControl.CustomMessage.class.getSimpleName();
+ }
+ public java.lang.String messageFullName()
+ {
+ return org.apache.drill.exec.proto.BitControl.CustomMessage.class.getName();
+ }
+ //unused
+ public void mergeFrom(com.dyuproject.protostuff.Input input, org.apache.drill.exec.proto.BitControl.CustomMessage message) throws java.io.IOException {}
+ public org.apache.drill.exec.proto.BitControl.CustomMessage newMessage() { return null; }
+ }
+ public static class BuilderSchema implements com.dyuproject.protostuff.Schema<org.apache.drill.exec.proto.BitControl.CustomMessage.Builder>
+ {
+ public void mergeFrom(com.dyuproject.protostuff.Input input, org.apache.drill.exec.proto.BitControl.CustomMessage.Builder builder) throws java.io.IOException
+ {
+ for(int number = input.readFieldNumber(this);; number = input.readFieldNumber(this))
+ {
+ switch(number)
+ {
+ case 0:
+ return;
+ case 1:
+ builder.setType(input.readInt32());
+ break;
+ case 2:
+ builder.setMessage(com.google.protobuf.ByteString.copyFrom(input.readByteArray()));
+ break;
+ default:
+ input.handleUnknownField(number, this);
+ }
+ }
+ }
+ public boolean isInitialized(org.apache.drill.exec.proto.BitControl.CustomMessage.Builder builder)
+ {
+ return builder.isInitialized();
+ }
+ public org.apache.drill.exec.proto.BitControl.CustomMessage.Builder newMessage()
+ {
+ return org.apache.drill.exec.proto.BitControl.CustomMessage.newBuilder();
+ }
+ public java.lang.String getFieldName(int number)
+ {
+ return org.apache.drill.exec.proto.SchemaBitControl.CustomMessage.getFieldName(number);
+ }
+ public int getFieldNumber(java.lang.String name)
+ {
+ return org.apache.drill.exec.proto.SchemaBitControl.CustomMessage.getFieldNumber(name);
+ }
+ public java.lang.Class<org.apache.drill.exec.proto.BitControl.CustomMessage.Builder> typeClass()
+ {
+ return org.apache.drill.exec.proto.BitControl.CustomMessage.Builder.class;
+ }
+ public java.lang.String messageName()
+ {
+ return org.apache.drill.exec.proto.BitControl.CustomMessage.class.getSimpleName();
+ }
+ public java.lang.String messageFullName()
+ {
+ return org.apache.drill.exec.proto.BitControl.CustomMessage.class.getName();
+ }
+ //unused
+ public void writeTo(com.dyuproject.protostuff.Output output, org.apache.drill.exec.proto.BitControl.CustomMessage.Builder builder) throws java.io.IOException {}
+ }
+ public static java.lang.String getFieldName(int number)
+ {
+ switch(number)
+ {
+ case 1: return "type";
+ case 2: return "message";
+ default: return null;
+ }
+ }
+ public static int getFieldNumber(java.lang.String name)
+ {
+ java.lang.Integer number = fieldMap.get(name);
+ return number == null ? 0 : number.intValue();
+ }
+ private static final java.util.HashMap<java.lang.String,java.lang.Integer> fieldMap = new java.util.HashMap<java.lang.String,java.lang.Integer>();
+ static
+ {
+ fieldMap.put("type", 1);
+ fieldMap.put("message", 2);
+ }
+ }
+
public static final class PlanFragment
{
public static final org.apache.drill.exec.proto.SchemaBitControl.PlanFragment.MessageSchema WRITE =
http://git-wip-us.apache.org/repos/asf/drill/blob/67d5cc6d/protocol/src/main/java/org/apache/drill/exec/proto/beans/CustomMessage.java
----------------------------------------------------------------------
diff --git a/protocol/src/main/java/org/apache/drill/exec/proto/beans/CustomMessage.java b/protocol/src/main/java/org/apache/drill/exec/proto/beans/CustomMessage.java
new file mode 100644
index 0000000..af9195e
--- /dev/null
+++ b/protocol/src/main/java/org/apache/drill/exec/proto/beans/CustomMessage.java
@@ -0,0 +1,186 @@
+/**
+ * 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.
+ */
+// Generated by http://code.google.com/p/protostuff/ ... DO NOT EDIT!
+// Generated from protobuf
+
+package org.apache.drill.exec.proto.beans;
+
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+
+import com.dyuproject.protostuff.ByteString;
+import com.dyuproject.protostuff.GraphIOUtil;
+import com.dyuproject.protostuff.Input;
+import com.dyuproject.protostuff.Message;
+import com.dyuproject.protostuff.Output;
+import com.dyuproject.protostuff.Schema;
+
+public final class CustomMessage implements Externalizable, Message<CustomMessage>, Schema<CustomMessage>
+{
+
+ public static Schema<CustomMessage> getSchema()
+ {
+ return DEFAULT_INSTANCE;
+ }
+
+ public static CustomMessage getDefaultInstance()
+ {
+ return DEFAULT_INSTANCE;
+ }
+
+ static final CustomMessage DEFAULT_INSTANCE = new CustomMessage();
+
+
+ private int type;
+ private ByteString message;
+
+ public CustomMessage()
+ {
+
+ }
+
+ // getters and setters
+
+ // type
+
+ public int getType()
+ {
+ return type;
+ }
+
+ public CustomMessage setType(int type)
+ {
+ this.type = type;
+ return this;
+ }
+
+ // message
+
+ public ByteString getMessage()
+ {
+ return message;
+ }
+
+ public CustomMessage setMessage(ByteString message)
+ {
+ this.message = message;
+ return this;
+ }
+
+ // java serialization
+
+ public void readExternal(ObjectInput in) throws IOException
+ {
+ GraphIOUtil.mergeDelimitedFrom(in, this, this);
+ }
+
+ public void writeExternal(ObjectOutput out) throws IOException
+ {
+ GraphIOUtil.writeDelimitedTo(out, this, this);
+ }
+
+ // message method
+
+ public Schema<CustomMessage> cachedSchema()
+ {
+ return DEFAULT_INSTANCE;
+ }
+
+ // schema methods
+
+ public CustomMessage newMessage()
+ {
+ return new CustomMessage();
+ }
+
+ public Class<CustomMessage> typeClass()
+ {
+ return CustomMessage.class;
+ }
+
+ public String messageName()
+ {
+ return CustomMessage.class.getSimpleName();
+ }
+
+ public String messageFullName()
+ {
+ return CustomMessage.class.getName();
+ }
+
+ public boolean isInitialized(CustomMessage message)
+ {
+ return true;
+ }
+
+ public void mergeFrom(Input input, CustomMessage message) throws IOException
+ {
+ for(int number = input.readFieldNumber(this);; number = input.readFieldNumber(this))
+ {
+ switch(number)
+ {
+ case 0:
+ return;
+ case 1:
+ message.type = input.readInt32();
+ break;
+ case 2:
+ message.message = input.readBytes();
+ break;
+ default:
+ input.handleUnknownField(number, this);
+ }
+ }
+ }
+
+
+ public void writeTo(Output output, CustomMessage message) throws IOException
+ {
+ if(message.type != 0)
+ output.writeInt32(1, message.type, false);
+
+ if(message.message != null)
+ output.writeBytes(2, message.message, false);
+ }
+
+ public String getFieldName(int number)
+ {
+ switch(number)
+ {
+ case 1: return "type";
+ case 2: return "message";
+ default: return null;
+ }
+ }
+
+ public int getFieldNumber(String name)
+ {
+ final Integer number = __fieldMap.get(name);
+ return number == null ? 0 : number.intValue();
+ }
+
+ private static final java.util.HashMap<String,Integer> __fieldMap = new java.util.HashMap<String,Integer>();
+ static
+ {
+ __fieldMap.put("type", 1);
+ __fieldMap.put("message", 2);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/67d5cc6d/protocol/src/main/protobuf/BitControl.proto
----------------------------------------------------------------------
diff --git a/protocol/src/main/protobuf/BitControl.proto b/protocol/src/main/protobuf/BitControl.proto
index 6d1b529..ca441f7 100644
--- a/protocol/src/main/protobuf/BitControl.proto
+++ b/protocol/src/main/protobuf/BitControl.proto
@@ -25,12 +25,14 @@ enum RpcType {
REQ_QUERY_STATUS = 10;
REQ_QUERY_CANCEL = 15;
REQ_UNPAUSE_FRAGMENT = 16; // send a resume message for a fragment, returns Ack
+ REQ_CUSTOM = 17;
// bit responses
RESP_FRAGMENT_HANDLE = 11;
RESP_FRAGMENT_STATUS = 12;
RESP_BIT_STATUS = 13;
RESP_QUERY_STATUS = 14;
+ RESP_CUSTOM = 18;
}
message BitControlHandshake{
@@ -52,6 +54,11 @@ message InitializeFragments {
repeated PlanFragment fragment = 1;
}
+message CustomMessage {
+ optional int32 type = 1;
+ optional bytes message = 2;
+}
+
message PlanFragment {
optional FragmentHandle handle = 1;
optional float network_cost = 4;
[3/3] drill git commit: DRILL-4257: Fix StoragePluginRegistry
clean-up behavior and misc clean up.
Posted by ja...@apache.org.
DRILL-4257: Fix StoragePluginRegistry clean-up behavior and misc clean up.
- Create the storage plugin registry using a Drill configuration parameter to be able to replace the registry implementation.
- Change StoragePluginRegistry into an interface and move the implementation to an impl class.
- Write documentation for StoragePluginRegistry.
- Make StoragePluginRegistry and StoragePluginMap AutoCloseable and ensure that Drillbit closes registry.
- Misc Drillbit code reorganization so that fields are at top of class (and static methods at bottom).
- Update DrillConfig to support reflection-based creation of storage plugin objects
- Remove final from DrillConfig so that application developers can extend config object.
This closes #321.
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/f964908a
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/f964908a
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/f964908a
Branch: refs/heads/master
Commit: f964908ae371b8321ce550a95add26d1c2c080a7
Parents: 67d5cc6
Author: Jacques Nadeau <ja...@apache.org>
Authored: Sat Jan 9 17:48:26 2016 -0800
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sun Jan 10 22:42:22 2016 -0800
----------------------------------------------------------------------
.../apache/drill/common/config/DrillConfig.java | 36 +-
.../org/apache/drill/exec/server/Drillbit.java | 217 ++++----
.../drill/exec/server/DrillbitContext.java | 10 +-
.../drill/exec/store/StoragePluginMap.java | 20 +-
.../drill/exec/store/StoragePluginRegistry.java | 500 +++----------------
.../exec/store/StoragePluginRegistryImpl.java | 497 ++++++++++++++++++
.../src/main/resources/drill-module.conf | 1 +
.../java/org/apache/drill/PlanningBase.java | 3 +-
.../drill/exec/memory/TestAllocators.java | 3 +-
.../exec/physical/impl/TestOptiqPlans.java | 3 +-
.../exec/physical/impl/join/TestMergeJoin.java | 17 +-
11 files changed, 762 insertions(+), 545 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/f964908a/common/src/main/java/org/apache/drill/common/config/DrillConfig.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/drill/common/config/DrillConfig.java b/common/src/main/java/org/apache/drill/common/config/DrillConfig.java
index 507c4f0..fa5dd29 100644
--- a/common/src/main/java/org/apache/drill/common/config/DrillConfig.java
+++ b/common/src/main/java/org/apache/drill/common/config/DrillConfig.java
@@ -19,6 +19,7 @@ package org.apache.drill.common.config;
import java.lang.management.ManagementFactory;
import java.lang.management.RuntimeMXBean;
+import java.lang.reflect.Constructor;
import java.net.URL;
import java.util.Collection;
import java.util.List;
@@ -27,17 +28,19 @@ import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.apache.drill.common.exceptions.DrillConfigurationException;
+import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.common.scanner.ClassPathScanner;
import org.reflections.util.ClasspathHelper;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import com.google.common.base.Stopwatch;
import com.google.common.collect.ImmutableList;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import com.typesafe.config.ConfigRenderOptions;
-public final class DrillConfig extends NestedConfig{
+public class DrillConfig extends NestedConfig {
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillConfig.class);
private final ImmutableList<String> startupArguments;
@@ -58,6 +61,36 @@ public final class DrillConfig extends NestedConfig{
logger.debug("DrillConfig object initialized.");
}
+ /**
+ * Get an instance of the provided interface using the configuration path provided. Construct the object based on the
+ * provided constructor arguments.
+ * @param path
+ * The configuration path to use.
+ * @param iface
+ * The Interface or Superclass of the instance you requested.
+ * @param constructorArgs
+ * Any arguments required for constructing the requested type.
+ * @return The new Object instance that implements the provided Interface
+ */
+ @SuppressWarnings("unchecked")
+ public <T> T getInstance(String path, Class<T> iface, Object... constructorArgs) {
+ try{
+ String className = this.getString(path);
+ Class<?> clazz = Class.forName(className);
+ Preconditions.checkArgument(iface.isAssignableFrom(clazz));
+ Class<?>[] argClasses = new Class[constructorArgs.length];
+ for (int i = 0; i < constructorArgs.length; i++) {
+ argClasses[i] = constructorArgs[i].getClass();
+ }
+ Constructor<?> constructor = clazz.getConstructor(argClasses);
+ return (T) constructor.newInstance(constructorArgs);
+ }catch(Exception e){
+ throw UserException.unsupportedError(e)
+ .message("Failure while attempting to load instance of the class of type %s requested at path %s.",
+ iface.getName(), path).build(logger);
+ }
+ }
+
public List<String> getStartupArguments() {
return startupArguments;
}
@@ -81,6 +114,7 @@ public final class DrillConfig extends NestedConfig{
return create(null, false);
}
+
/**
* <p>
* DrillConfig loads up Drill configuration information. It does this utilizing a combination of classpath scanning
http://git-wip-us.apache.org/repos/asf/drill/blob/f964908a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
index 99523d6..9734a38 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
@@ -36,6 +36,7 @@ import org.apache.drill.exec.server.options.OptionValue;
import org.apache.drill.exec.server.options.OptionValue.OptionType;
import org.apache.drill.exec.server.rest.WebServer;
import org.apache.drill.exec.service.ServiceEngine;
+import org.apache.drill.exec.store.StoragePluginRegistry;
import org.apache.drill.exec.store.sys.CachingStoreProvider;
import org.apache.drill.exec.store.sys.PStoreProvider;
import org.apache.drill.exec.store.sys.PStoreRegistry;
@@ -51,117 +52,14 @@ import com.google.common.base.Stopwatch;
*/
public class Drillbit implements AutoCloseable {
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(Drillbit.class);
+
static {
Environment.logEnv("Drillbit environment: ", logger);
}
- private boolean isClosed = false;
-
- public static Drillbit start(final StartupOptions options) throws DrillbitStartupException {
- return start(DrillConfig.create(options.getConfigLocation()), null);
- }
-
- public static Drillbit start(final DrillConfig config) throws DrillbitStartupException {
- return start(config, null);
- }
-
- public static Drillbit start(final DrillConfig config, final RemoteServiceSet remoteServiceSet)
- throws DrillbitStartupException {
- logger.debug("Starting new Drillbit.");
- // TODO: allow passing as a parameter
- ScanResult classpathScan = ClassPathScanner.fromPrescan(config);
- Drillbit bit;
- try {
- bit = new Drillbit(config, remoteServiceSet, classpathScan);
- } catch (final Exception ex) {
- throw new DrillbitStartupException("Failure while initializing values in Drillbit.", ex);
- }
-
- try {
- bit.run();
- } catch (final Exception e) {
- bit.close();
- throw new DrillbitStartupException("Failure during initial startup of Drillbit.", e);
- }
- logger.debug("Started new Drillbit.");
- return bit;
- }
-
private final static String SYSTEM_OPTIONS_NAME = "org.apache.drill.exec.server.Drillbit.system_options";
- private static void throwInvalidSystemOption(final String systemProp, final String errorMessage) {
- throw new IllegalStateException("Property \"" + SYSTEM_OPTIONS_NAME + "\" part \"" + systemProp
- + "\" " + errorMessage + ".");
- }
-
- private static String stripQuotes(final String s, final String systemProp) {
- if (s.isEmpty()) {
- return s;
- }
-
- final char cFirst = s.charAt(0);
- final char cLast = s.charAt(s.length() - 1);
- if ((cFirst == '"') || (cFirst == '\'')) {
- if (cLast != cFirst) {
- throwInvalidSystemOption(systemProp, "quoted value does not have closing quote");
- }
-
- return s.substring(1, s.length() - 2); // strip the quotes
- }
-
- if ((cLast == '"') || (cLast == '\'')) {
- throwInvalidSystemOption(systemProp, "value has unbalanced closing quote");
- }
-
- // return as-is
- return s;
- }
-
- private void javaPropertiesToSystemOptions() {
- // get the system options property
- final String allSystemProps = System.getProperty(SYSTEM_OPTIONS_NAME);
- if ((allSystemProps == null) || allSystemProps.isEmpty()) {
- return;
- }
-
- final OptionManager optionManager = getContext().getOptionManager();
-
- // parse out the properties, validate, and then set them
- final String systemProps[] = allSystemProps.split(",");
- for(final String systemProp : systemProps) {
- final String keyValue[] = systemProp.split("=");
- if (keyValue.length != 2) {
- throwInvalidSystemOption(systemProp, "does not contain a key=value assignment");
- }
-
- final String optionName = keyValue[0].trim();
- if (optionName.isEmpty()) {
- throwInvalidSystemOption(systemProp, "does not contain a key before the assignment");
- }
-
- final String optionString = stripQuotes(keyValue[1].trim(), systemProp);
- if (optionString.isEmpty()) {
- throwInvalidSystemOption(systemProp, "does not contain a value after the assignment");
- }
-
- final OptionValue defaultValue = optionManager.getOption(optionName);
- if (defaultValue == null) {
- throwInvalidSystemOption(systemProp, "does not specify a valid option name");
- }
- if (defaultValue.type != OptionType.SYSTEM) {
- throwInvalidSystemOption(systemProp, "does not specify a SYSTEM option ");
- }
-
- final OptionValue optionValue = OptionValue.createOption(
- defaultValue.kind, OptionType.SYSTEM, optionName, optionString);
- optionManager.setOption(optionValue);
- }
- }
-
- public static void main(final String[] cli) throws DrillbitStartupException {
- final StartupOptions options = StartupOptions.parse(cli);
- start(options);
- }
+ private boolean isClosed = false;
private final ClusterCoordinator coord;
private final ServiceEngine engine;
@@ -170,6 +68,7 @@ public class Drillbit implements AutoCloseable {
private final BootStrapContext context;
private final WebServer webServer;
private RegistrationHandle registrationHandle;
+ private volatile StoragePluginRegistry storageRegistry;
@VisibleForTesting
public Drillbit(
@@ -210,7 +109,8 @@ public class Drillbit implements AutoCloseable {
final DrillbitEndpoint md = engine.start();
manager.start(md, engine.getController(), engine.getDataConnectionCreator(), coord, storeProvider);
final DrillbitContext drillbitContext = manager.getContext();
- drillbitContext.getStorage().init();
+ storageRegistry = drillbitContext.getStorage();
+ storageRegistry.init();
drillbitContext.getOptionManager().init();
javaPropertiesToSystemOptions();
registrationHandle = coord.register(md);
@@ -252,6 +152,7 @@ public class Drillbit implements AutoCloseable {
storeProvider,
coord,
manager,
+ storageRegistry,
context);
} catch(Exception e) {
logger.warn("Failure on close()", e);
@@ -261,6 +162,47 @@ public class Drillbit implements AutoCloseable {
isClosed = true;
}
+ private void javaPropertiesToSystemOptions() {
+ // get the system options property
+ final String allSystemProps = System.getProperty(SYSTEM_OPTIONS_NAME);
+ if ((allSystemProps == null) || allSystemProps.isEmpty()) {
+ return;
+ }
+
+ final OptionManager optionManager = getContext().getOptionManager();
+
+ // parse out the properties, validate, and then set them
+ final String systemProps[] = allSystemProps.split(",");
+ for (final String systemProp : systemProps) {
+ final String keyValue[] = systemProp.split("=");
+ if (keyValue.length != 2) {
+ throwInvalidSystemOption(systemProp, "does not contain a key=value assignment");
+ }
+
+ final String optionName = keyValue[0].trim();
+ if (optionName.isEmpty()) {
+ throwInvalidSystemOption(systemProp, "does not contain a key before the assignment");
+ }
+
+ final String optionString = stripQuotes(keyValue[1].trim(), systemProp);
+ if (optionString.isEmpty()) {
+ throwInvalidSystemOption(systemProp, "does not contain a value after the assignment");
+ }
+
+ final OptionValue defaultValue = optionManager.getOption(optionName);
+ if (defaultValue == null) {
+ throwInvalidSystemOption(systemProp, "does not specify a valid option name");
+ }
+ if (defaultValue.type != OptionType.SYSTEM) {
+ throwInvalidSystemOption(systemProp, "does not specify a SYSTEM option ");
+ }
+
+ final OptionValue optionValue = OptionValue.createOption(
+ defaultValue.kind, OptionType.SYSTEM, optionName, optionString);
+ optionManager.setOption(optionValue);
+ }
+ }
+
/**
* Shutdown hook for Drillbit. Closes the drillbit, and reports on errors that
* occur during closure, as well as the location the drillbit was started from.
@@ -310,4 +252,67 @@ public class Drillbit implements AutoCloseable {
return manager.getContext();
}
+ public static void main(final String[] cli) throws DrillbitStartupException {
+ final StartupOptions options = StartupOptions.parse(cli);
+ start(options);
+ }
+
+ public static Drillbit start(final StartupOptions options) throws DrillbitStartupException {
+ return start(DrillConfig.create(options.getConfigLocation()), null);
+ }
+
+ public static Drillbit start(final DrillConfig config) throws DrillbitStartupException {
+ return start(config, null);
+ }
+
+ public static Drillbit start(final DrillConfig config, final RemoteServiceSet remoteServiceSet)
+ throws DrillbitStartupException {
+ logger.debug("Starting new Drillbit.");
+ // TODO: allow passing as a parameter
+ ScanResult classpathScan = ClassPathScanner.fromPrescan(config);
+ Drillbit bit;
+ try {
+ bit = new Drillbit(config, remoteServiceSet, classpathScan);
+ } catch (final Exception ex) {
+ throw new DrillbitStartupException("Failure while initializing values in Drillbit.", ex);
+ }
+
+ try {
+ bit.run();
+ } catch (final Exception e) {
+ bit.close();
+ throw new DrillbitStartupException("Failure during initial startup of Drillbit.", e);
+ }
+ logger.debug("Started new Drillbit.");
+ return bit;
+ }
+
+ private static void throwInvalidSystemOption(final String systemProp, final String errorMessage) {
+ throw new IllegalStateException("Property \"" + SYSTEM_OPTIONS_NAME + "\" part \"" + systemProp
+ + "\" " + errorMessage + ".");
+ }
+
+ private static String stripQuotes(final String s, final String systemProp) {
+ if (s.isEmpty()) {
+ return s;
+ }
+
+ final char cFirst = s.charAt(0);
+ final char cLast = s.charAt(s.length() - 1);
+ if ((cFirst == '"') || (cFirst == '\'')) {
+ if (cLast != cFirst) {
+ throwInvalidSystemOption(systemProp, "quoted value does not have closing quote");
+ }
+
+ return s.substring(1, s.length() - 2); // strip the quotes
+ }
+
+ if ((cLast == '"') || (cLast == '\'')) {
+ throwInvalidSystemOption(systemProp, "value has unbalanced closing quote");
+ }
+
+ // return as-is
+ return s;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/f964908a/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
index c1f2e5b..aa6a0da 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
@@ -37,8 +37,8 @@ import org.apache.drill.exec.rpc.control.Controller;
import org.apache.drill.exec.rpc.control.WorkEventBus;
import org.apache.drill.exec.rpc.data.DataConnectionCreator;
import org.apache.drill.exec.server.options.SystemOptionManager;
+import org.apache.drill.exec.store.SchemaFactory;
import org.apache.drill.exec.store.StoragePluginRegistry;
-import org.apache.drill.exec.store.StoragePluginRegistry.DrillSchemaFactory;
import org.apache.drill.exec.store.sys.PStoreProvider;
import com.codahale.metrics.MetricRegistry;
@@ -80,7 +80,11 @@ public class DrillbitContext {
this.endpoint = checkNotNull(endpoint);
this.provider = provider;
this.lpPersistence = new LogicalPlanPersistence(context.getConfig(), classpathScan);
- this.storagePlugins = new StoragePluginRegistry(this); // TODO change constructor
+
+ // TODO remove escaping "this".
+ this.storagePlugins = context.getConfig()
+ .getInstance(StoragePluginRegistry.STORAGE_PLUGIN_REGISTRY_IMPL, StoragePluginRegistry.class, this);
+
this.reader = new PhysicalPlanReader(context.getConfig(), classpathScan, lpPersistence, endpoint, storagePlugins);
this.operatorCreatorRegistry = new OperatorCreatorRegistry(classpathScan);
this.systemOptions = new SystemOptionManager(lpPersistence, provider);
@@ -152,7 +156,7 @@ public class DrillbitContext {
return provider;
}
- public DrillSchemaFactory getSchemaFactory() {
+ public SchemaFactory getSchemaFactory() {
return storagePlugins.getSchemaFactory();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/f964908a/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginMap.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginMap.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginMap.java
index f12d195..2f4a929 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginMap.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginMap.java
@@ -23,6 +23,7 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentMap;
+import org.apache.drill.common.AutoCloseables;
import org.apache.drill.common.logical.StoragePluginConfig;
import com.google.common.collect.LinkedListMultimap;
@@ -36,7 +37,7 @@ import com.google.common.collect.Multimaps;
* This is inspired by ConcurrentMap but provides a secondary key mapping that allows an alternative lookup mechanism.
* The class is responsible for internally managing consistency between the two maps. This class is threadsafe.
*/
-class StoragePluginMap implements Iterable<Entry<String, StoragePlugin>> {
+class StoragePluginMap implements Iterable<Entry<String, StoragePlugin>>, AutoCloseable {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(StoragePluginMap.class);
private final ConcurrentMap<String, StoragePlugin> nameMap = Maps.newConcurrentMap();
@@ -73,6 +74,18 @@ class StoragePluginMap implements Iterable<Entry<String, StoragePlugin>> {
return ok;
}
+ public void put(String name, StoragePlugin plugin) {
+ StoragePlugin oldPlugin = nameMap.put(name, plugin);
+ configMap.put(plugin.getConfig(), plugin);
+ if (oldPlugin != null) {
+ try {
+ oldPlugin.close();
+ } catch (Exception e) {
+ logger.warn("Failure while closing plugin replaced by injection.", e);
+ }
+ }
+ }
+
public StoragePlugin putIfAbsent(String name, StoragePlugin plugin) {
StoragePlugin oldPlugin = nameMap.putIfAbsent(name, plugin);
if (oldPlugin == null) {
@@ -117,4 +130,9 @@ class StoragePluginMap implements Iterable<Entry<String, StoragePlugin>> {
return nameMap.values();
}
+ @Override
+ public void close() throws Exception {
+ AutoCloseables.close(configMap.values());
+ }
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/f964908a/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginRegistry.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginRegistry.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginRegistry.java
index 9d7dc57..c7d364b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginRegistry.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginRegistry.java
@@ -17,455 +17,107 @@
*/
package org.apache.drill.exec.store;
-import static com.google.common.base.Preconditions.checkNotNull;
-
-import java.io.IOException;
-import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
-import java.net.URL;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import org.apache.calcite.plan.RelOptRule;
-import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.tools.RuleSet;
-import org.apache.drill.common.config.LogicalPlanPersistence;
-import org.apache.drill.common.exceptions.DrillRuntimeException;
import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.common.logical.FormatPluginConfig;
import org.apache.drill.common.logical.StoragePluginConfig;
-import org.apache.drill.common.scanner.ClassPathScanner;
-import org.apache.drill.common.scanner.persistence.ScanResult;
-import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.exception.DrillbitStartupException;
import org.apache.drill.exec.ops.OptimizerRulesContext;
-import org.apache.drill.exec.planner.logical.DrillRuleSets;
-import org.apache.drill.exec.planner.logical.StoragePlugins;
-import org.apache.drill.exec.server.DrillbitContext;
-import org.apache.drill.exec.store.dfs.FileSystemPlugin;
import org.apache.drill.exec.store.dfs.FormatPlugin;
-import org.apache.drill.exec.store.ischema.InfoSchemaConfig;
-import org.apache.drill.exec.store.ischema.InfoSchemaStoragePlugin;
import org.apache.drill.exec.store.sys.PStore;
-import org.apache.drill.exec.store.sys.PStoreConfig;
-import org.apache.drill.exec.store.sys.SystemTablePlugin;
-import org.apache.drill.exec.store.sys.SystemTablePluginConfig;
-
-import com.google.common.base.Charsets;
-import com.google.common.base.Joiner;
-import com.google.common.base.Stopwatch;
-import com.google.common.cache.CacheBuilder;
-import com.google.common.cache.CacheLoader;
-import com.google.common.cache.LoadingCache;
-import com.google.common.cache.RemovalListener;
-import com.google.common.cache.RemovalNotification;
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.ImmutableSet.Builder;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-import com.google.common.io.Resources;
-
-public class StoragePluginRegistry implements Iterable<Map.Entry<String, StoragePlugin>> {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(StoragePluginRegistry.class);
-
- public static final String SYS_PLUGIN = "sys";
-
- public static final String INFORMATION_SCHEMA_PLUGIN = "INFORMATION_SCHEMA";
-
- private Map<Object, Constructor<? extends StoragePlugin>> availablePlugins = new HashMap<Object, Constructor<? extends StoragePlugin>>();
- private final StoragePluginMap plugins = new StoragePluginMap();
-
- private DrillbitContext context;
- private final DrillSchemaFactory schemaFactory = new DrillSchemaFactory();
- private final PStore<StoragePluginConfig> pluginSystemTable;
- private final LogicalPlanPersistence lpPersistence;
- private final ScanResult classpathScan;
- private final LoadingCache<StoragePluginConfig, StoragePlugin> ephemeralPlugins;
-
- public StoragePluginRegistry(DrillbitContext context) {
- this.context = checkNotNull(context);
- this.lpPersistence = checkNotNull(context.getLpPersistence());
- this.classpathScan = checkNotNull(context.getClasspathScan());
- try {
- this.pluginSystemTable = context //
- .getPersistentStoreProvider() //
- .getStore(PStoreConfig //
- .newJacksonBuilder(lpPersistence.getMapper(), StoragePluginConfig.class) //
- .name("sys.storage_plugins") //
- .build());
- } catch (IOException | RuntimeException e) {
- logger.error("Failure while loading storage plugin registry.", e);
- throw new RuntimeException("Failure while reading and loading storage plugin configuration.", e);
- }
-
- ephemeralPlugins = CacheBuilder.newBuilder()
- .expireAfterAccess(24, TimeUnit.HOURS)
- .maximumSize(250)
- .removalListener(new RemovalListener<StoragePluginConfig, StoragePlugin>() {
- @Override
- public void onRemoval(RemovalNotification<StoragePluginConfig, StoragePlugin> notification) {
- closePlugin(notification.getValue());
- }
- })
- .build(new CacheLoader<StoragePluginConfig, StoragePlugin>() {
- @Override
- public StoragePlugin load(StoragePluginConfig config) throws Exception {
- return create(null, config);
- }
- });
- }
-
- public PStore<StoragePluginConfig> getStore() {
- return pluginSystemTable;
- }
-
- @SuppressWarnings("unchecked")
- public void init() throws DrillbitStartupException {
- final Collection<Class<? extends StoragePlugin>> pluginClasses =
- classpathScan.getImplementations(StoragePlugin.class);
- final String lineBrokenList =
- pluginClasses.size() == 0
- ? "" : "\n\t- " + Joiner.on("\n\t- ").join(pluginClasses);
- logger.debug("Found {} storage plugin configuration classes: {}.",
- pluginClasses.size(), lineBrokenList);
- for (Class<? extends StoragePlugin> plugin : pluginClasses) {
- int i = 0;
- for (Constructor<?> c : plugin.getConstructors()) {
- Class<?>[] params = c.getParameterTypes();
- if(params.length != 3
- || params[1] != DrillbitContext.class
- || !StoragePluginConfig.class.isAssignableFrom(params[0])
- || params[2] != String.class) {
- logger.info("Skipping StoragePlugin constructor {} for plugin class {} since it doesn't implement a "
- + "[constructor(StoragePluginConfig, DrillbitContext, String)]", c, plugin);
- continue;
- }
- availablePlugins.put(params[0], (Constructor<? extends StoragePlugin>) c);
- i++;
- }
- if (i == 0) {
- logger.debug("Skipping registration of StoragePlugin {} as it doesn't have a constructor with the parameters "
- + "of (StorangePluginConfig, Config)", plugin.getCanonicalName());
- }
- }
-
- // create registered plugins defined in "storage-plugins.json"
- this.plugins.putAll(createPlugins());
-
- }
- private Map<String, StoragePlugin> createPlugins() throws DrillbitStartupException {
- try {
- /*
- * Check if the storage plugins system table has any entries. If not, load the boostrap-storage-plugin file into the system table.
- */
- if (!pluginSystemTable.iterator().hasNext()) {
- // bootstrap load the config since no plugins are stored.
- logger.info("No storage plugin instances configured in persistent store, loading bootstrap configuration.");
- Collection<URL> urls = ClassPathScanner.forResource(ExecConstants.BOOTSTRAP_STORAGE_PLUGINS_FILE, false);
- if (urls != null && ! urls.isEmpty()) {
- logger.info("Loading the storage plugin configs from URLs {}.", urls);
- Map<String, URL> pluginURLMap = Maps.newHashMap();
- for (URL url :urls) {
- String pluginsData = Resources.toString(url, Charsets.UTF_8);
- StoragePlugins plugins = lpPersistence.getMapper().readValue(pluginsData, StoragePlugins.class);
- for (Map.Entry<String, StoragePluginConfig> config : plugins) {
- if (!pluginSystemTable.putIfAbsent(config.getKey(), config.getValue())) {
- logger.warn("Duplicate plugin instance '{}' defined in [{}, {}], ignoring the later one.",
- config.getKey(), pluginURLMap.get(config.getKey()), url);
- continue;
- }
- pluginURLMap.put(config.getKey(), url);
- }
- }
- } else {
- throw new IOException("Failure finding " + ExecConstants.BOOTSTRAP_STORAGE_PLUGINS_FILE);
- }
- }
+public interface StoragePluginRegistry extends Iterable<Map.Entry<String, StoragePlugin>>, AutoCloseable {
+ final String SYS_PLUGIN = "sys";
+ final String INFORMATION_SCHEMA_PLUGIN = "INFORMATION_SCHEMA";
+ final String STORAGE_PLUGIN_REGISTRY_IMPL = "drill.exec.storage.registry";
+ final String PSTORE_NAME = "sys.storage_plugins";
- Map<String, StoragePlugin> activePlugins = new HashMap<String, StoragePlugin>();
- for (Map.Entry<String, StoragePluginConfig> entry : pluginSystemTable) {
- String name = entry.getKey();
- StoragePluginConfig config = entry.getValue();
- if (config.isEnabled()) {
- try {
- StoragePlugin plugin = create(name, config);
- activePlugins.put(name, plugin);
- } catch (ExecutionSetupException e) {
- logger.error("Failure while setting up StoragePlugin with name: '{}', disabling.", name, e);
- config.setEnabled(false);
- pluginSystemTable.put(name, config);
- }
- }
- }
-
- activePlugins.put(INFORMATION_SCHEMA_PLUGIN, new InfoSchemaStoragePlugin(new InfoSchemaConfig(), context, INFORMATION_SCHEMA_PLUGIN));
- activePlugins.put(SYS_PLUGIN, new SystemTablePlugin(SystemTablePluginConfig.INSTANCE, context, SYS_PLUGIN));
-
- return activePlugins;
- } catch (IOException e) {
- logger.error("Failure setting up storage plugins. Drillbit exiting.", e);
- throw new IllegalStateException(e);
- }
- }
-
- public void deletePlugin(String name) {
- StoragePlugin plugin = plugins.remove(name);
- closePlugin(plugin);
- pluginSystemTable.delete(name);
- }
-
- private void closePlugin(StoragePlugin plugin) {
- if (plugin == null) {
- return;
- }
-
- try {
- plugin.close();
- } catch (Exception e) {
- logger.warn("Exception while shutting down storage plugin.");
- }
- }
-
- public StoragePlugin createOrUpdate(String name, StoragePluginConfig config, boolean persist) throws ExecutionSetupException {
- for (;;) {
- final StoragePlugin oldPlugin = plugins.get(name);
- final StoragePlugin newPlugin = create(name, config);
- boolean done = false;
- try {
- if (oldPlugin != null) {
- if (config.isEnabled()) {
- done = plugins.replace(name, oldPlugin, newPlugin);
- } else {
- done = plugins.remove(name, oldPlugin);
- }
- if (done) {
- closePlugin(oldPlugin);
- }
- } else if (config.isEnabled()) {
- done = (null == plugins.putIfAbsent(name, newPlugin));
- } else {
- done = true;
- }
- } finally {
- if (!done) {
- closePlugin(newPlugin);
- }
- }
-
- if (done) {
- if (persist) {
- pluginSystemTable.put(name, config);
- }
-
- return newPlugin;
- }
- }
- }
+ /**
+ * Initialize the storage plugin registry. Must be called before the registry is used.
+ *
+ * @throws DrillbitStartupException
+ */
+ void init() throws DrillbitStartupException;
- public StoragePlugin getPlugin(String name) throws ExecutionSetupException {
- StoragePlugin plugin = plugins.get(name);
- if (name.equals(SYS_PLUGIN) || name.equals(INFORMATION_SCHEMA_PLUGIN)) {
- return plugin;
- }
+ /**
+ * Delete a plugin by name
+ * @param name
+ * The name of the storage plugin to delete.
+ */
+ void deletePlugin(String name);
- // since we lazily manage the list of plugins per server, we need to update this once we know that it is time.
- StoragePluginConfig config = this.pluginSystemTable.get(name);
- if (config == null) {
- if (plugin != null) {
- plugins.remove(name);
- }
- return null;
- } else {
- if (plugin == null
- || !plugin.getConfig().equals(config)
- || plugin.getConfig().isEnabled() != config.isEnabled()) {
- plugin = createOrUpdate(name, config, false);
- }
- return plugin;
- }
- }
+ /**
+ * Create a plugin by name and configuration. If the plugin already exists, update the plugin
+ * @param name
+ * The name of the plugin
+ * @param config
+ * The plugin confgiruation
+ * @param persist
+ * Whether to persist the plugin for later use or treat it as ephemeral.
+ * @return The StoragePlugin instance.
+ * @throws ExecutionSetupException
+ */
+ StoragePlugin createOrUpdate(String name, StoragePluginConfig config, boolean persist) throws ExecutionSetupException;
- public StoragePlugin getPlugin(StoragePluginConfig config) throws ExecutionSetupException {
- if (config instanceof NamedStoragePluginConfig) {
- return getPlugin(((NamedStoragePluginConfig) config).name);
- } else {
- // try to lookup plugin by configuration
- StoragePlugin plugin = plugins.get(config);
- if (plugin != null) {
- return plugin;
- }
+ /**
+ * Get a plugin by name. Create it based on the PStore saved definition if it doesn't exist.
+ * @param name
+ * The name of the plugin
+ * @return The StoragePlugin instance.
+ * @throws ExecutionSetupException
+ */
+ StoragePlugin getPlugin(String name) throws ExecutionSetupException;
- // no named plugin matches the desired configuration, let's create an
- // ephemeral storage plugin (or get one from the cache)
- try {
- return ephemeralPlugins.get(config);
- } catch (ExecutionException e) {
- Throwable cause = e.getCause();
- if (cause instanceof ExecutionSetupException) {
- throw (ExecutionSetupException) cause;
- } else {
- // this shouldn't happen. here for completeness.
- throw new ExecutionSetupException("Failure while trying to create ephemeral plugin.", cause);
- }
- }
- }
- }
+ /**
+ * Get a plugin by configuration. If it doesn't exist, create it.
+ * @param config
+ * The configuration for the plugin.
+ * @return The StoragePlugin instance.
+ * @throws ExecutionSetupException
+ */
+ StoragePlugin getPlugin(StoragePluginConfig config) throws ExecutionSetupException;
- public FormatPlugin getFormatPlugin(StoragePluginConfig storageConfig, FormatPluginConfig formatConfig) throws ExecutionSetupException {
- StoragePlugin p = getPlugin(storageConfig);
- if (!(p instanceof FileSystemPlugin)) {
- throw new ExecutionSetupException(String.format("You tried to request a format plugin for a storage plugin that wasn't of type FileSystemPlugin. The actual type of plugin was %s.", p.getClass().getName()));
- }
- FileSystemPlugin storage = (FileSystemPlugin) p;
- return storage.getFormatPlugin(formatConfig);
- }
+ /**
+ * Add a plugin to the registry using the provided name.
+ *
+ * @param name
+ * @param plugin
+ */
+ void addPlugin(String name, StoragePlugin plugin);
- private StoragePlugin create(String name, StoragePluginConfig pluginConfig) throws ExecutionSetupException {
- StoragePlugin plugin = null;
- Constructor<? extends StoragePlugin> c = availablePlugins.get(pluginConfig.getClass());
- if (c == null) {
- throw new ExecutionSetupException(String.format("Failure finding StoragePlugin constructor for config %s",
- pluginConfig));
- }
- try {
- plugin = c.newInstance(pluginConfig, context, name);
- plugin.start();
- return plugin;
- } catch (InstantiationException | IllegalAccessException | IllegalArgumentException | InvocationTargetException
- | IOException e) {
- Throwable t = e instanceof InvocationTargetException ? ((InvocationTargetException) e).getTargetException() : e;
- if (t instanceof ExecutionSetupException) {
- throw ((ExecutionSetupException) t);
- }
- throw new ExecutionSetupException(String.format(
- "Failure setting up new storage plugin configuration for config %s", pluginConfig), t);
- }
- }
+ /**
+ * Get the Format plugin for the FileSystemPlugin associated with the provided storage config and format config.
+ *
+ * @param storageConfig
+ * The storage config for the associated FileSystemPlugin
+ * @param formatConfig
+ * The format config for the associated FormatPlugin
+ * @return A FormatPlugin
+ * @throws ExecutionSetupException
+ */
+ FormatPlugin getFormatPlugin(StoragePluginConfig storageConfig, FormatPluginConfig formatConfig)
+ throws ExecutionSetupException;
- @Override
- public Iterator<Entry<String, StoragePlugin>> iterator() {
- return plugins.iterator();
- }
+ /**
+ * Get the PStore for this StoragePluginRegistry. (Used in the management layer.)
+ * @return PStore for StoragePlugin configuration objects.
+ */
+ PStore<StoragePluginConfig> getStore();
/**
* Return StoragePlugin rule sets.
+ *
* @param optimizerRulesContext
* @return Array of logical and physical rule sets.
*/
- public RuleSet[] getStoragePluginRuleSet(OptimizerRulesContext optimizerRulesContext) {
- // query registered engines for optimizer rules and build the storage plugin RuleSet
- Builder<RelOptRule> logicalRulesBuilder = ImmutableSet.builder();
- Builder<RelOptRule> physicalRulesBuilder = ImmutableSet.builder();
- for (StoragePlugin plugin : this.plugins.plugins()) {
- Set<? extends RelOptRule> rules = plugin.getLogicalOptimizerRules(optimizerRulesContext);
- if (rules != null && rules.size() > 0) {
- logicalRulesBuilder.addAll(rules);
- }
- rules = plugin.getPhysicalOptimizerRules(optimizerRulesContext);
- if (rules != null && rules.size() > 0) {
- physicalRulesBuilder.addAll(rules);
- }
- }
-
- return new RuleSet[] {
- DrillRuleSets.create(logicalRulesBuilder.build()),
- DrillRuleSets.create(physicalRulesBuilder.build()) };
- }
-
- public DrillSchemaFactory getSchemaFactory() {
- return schemaFactory;
- }
-
- public class DrillSchemaFactory implements SchemaFactory {
-
- @Override
- public void registerSchemas(SchemaConfig schemaConfig, SchemaPlus parent) throws IOException {
- Stopwatch watch = new Stopwatch();
- watch.start();
-
- try {
- Set<String> currentPluginNames = Sets.newHashSet(plugins.names());
- // iterate through the plugin instances in the persistence store adding
- // any new ones and refreshing those whose configuration has changed
- for (Map.Entry<String, StoragePluginConfig> config : pluginSystemTable) {
- if (config.getValue().isEnabled()) {
- getPlugin(config.getKey());
- currentPluginNames.remove(config.getKey());
- }
- }
- // remove those which are no longer in the registry
- for (String pluginName : currentPluginNames) {
- if (pluginName.equals(SYS_PLUGIN) || pluginName.equals(INFORMATION_SCHEMA_PLUGIN)) {
- continue;
- }
- plugins.remove(pluginName);
- }
-
- // finally register schemas with the refreshed plugins
- for (StoragePlugin plugin : plugins.plugins()) {
- plugin.registerSchemas(schemaConfig, parent);
- }
- } catch (ExecutionSetupException e) {
- throw new DrillRuntimeException("Failure while updating storage plugins", e);
- }
-
- // Add second level schema as top level schema with name qualified with parent schema name
- // Ex: "dfs" schema has "default" and "tmp" as sub schemas. Add following extra schemas "dfs.default" and
- // "dfs.tmp" under root schema.
- //
- // Before change, schema tree looks like below:
- // "root"
- // -- "dfs"
- // -- "default"
- // -- "tmp"
- // -- "hive"
- // -- "default"
- // -- "hivedb1"
- //
- // After the change, the schema tree looks like below:
- // "root"
- // -- "dfs"
- // -- "default"
- // -- "tmp"
- // -- "dfs.default"
- // -- "dfs.tmp"
- // -- "hive"
- // -- "default"
- // -- "hivedb1"
- // -- "hive.default"
- // -- "hive.hivedb1"
- List<SchemaPlus> secondLevelSchemas = Lists.newArrayList();
- for (String firstLevelSchemaName : parent.getSubSchemaNames()) {
- SchemaPlus firstLevelSchema = parent.getSubSchema(firstLevelSchemaName);
- for (String secondLevelSchemaName : firstLevelSchema.getSubSchemaNames()) {
- secondLevelSchemas.add(firstLevelSchema.getSubSchema(secondLevelSchemaName));
- }
- }
-
- for (SchemaPlus schema : secondLevelSchemas) {
- AbstractSchema drillSchema;
- try {
- drillSchema = schema.unwrap(AbstractSchema.class);
- } catch (ClassCastException e) {
- throw new RuntimeException(String.format("Schema '%s' is not expected under root schema", schema.getName()));
- }
- SubSchemaWrapper wrapper = new SubSchemaWrapper(drillSchema);
- parent.add(wrapper.getName(), wrapper);
- }
-
- logger.debug("Took {} ms to register schemas.", watch.elapsed(TimeUnit.MILLISECONDS));
- }
-
- }
+ RuleSet[] getStoragePluginRuleSet(OptimizerRulesContext optimizerRulesContext);
+ /**
+ * Get the Schema factory associated with this storage plugin registry.
+ * @return A SchemaFactory that can register the schemas associated with this plugin registry.
+ */
+ SchemaFactory getSchemaFactory();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/f964908a/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginRegistryImpl.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginRegistryImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginRegistryImpl.java
new file mode 100644
index 0000000..a8628c7
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginRegistryImpl.java
@@ -0,0 +1,497 @@
+/**
+ * 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.drill.exec.store;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.net.URL;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.tools.RuleSet;
+import org.apache.drill.common.config.LogicalPlanPersistence;
+import org.apache.drill.common.exceptions.DrillRuntimeException;
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.logical.FormatPluginConfig;
+import org.apache.drill.common.logical.StoragePluginConfig;
+import org.apache.drill.common.scanner.ClassPathScanner;
+import org.apache.drill.common.scanner.persistence.ScanResult;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.exception.DrillbitStartupException;
+import org.apache.drill.exec.ops.OptimizerRulesContext;
+import org.apache.drill.exec.planner.logical.DrillRuleSets;
+import org.apache.drill.exec.planner.logical.StoragePlugins;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.store.dfs.FileSystemPlugin;
+import org.apache.drill.exec.store.dfs.FormatPlugin;
+import org.apache.drill.exec.store.ischema.InfoSchemaConfig;
+import org.apache.drill.exec.store.ischema.InfoSchemaStoragePlugin;
+import org.apache.drill.exec.store.sys.PStore;
+import org.apache.drill.exec.store.sys.PStoreConfig;
+import org.apache.drill.exec.store.sys.SystemTablePlugin;
+import org.apache.drill.exec.store.sys.SystemTablePluginConfig;
+
+import com.google.common.base.Charsets;
+import com.google.common.base.Joiner;
+import com.google.common.base.Stopwatch;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.cache.RemovalListener;
+import com.google.common.cache.RemovalNotification;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.ImmutableSet.Builder;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.common.io.Resources;
+
+public class StoragePluginRegistryImpl implements StoragePluginRegistry {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(StoragePluginRegistryImpl.class);
+
+ private Map<Object, Constructor<? extends StoragePlugin>> availablePlugins = Collections.emptyMap();
+ private final StoragePluginMap plugins = new StoragePluginMap();
+
+ private DrillbitContext context;
+ private final DrillSchemaFactory schemaFactory = new DrillSchemaFactory();
+ private final PStore<StoragePluginConfig> pluginSystemTable;
+ private final LogicalPlanPersistence lpPersistence;
+ private final ScanResult classpathScan;
+ private final LoadingCache<StoragePluginConfig, StoragePlugin> ephemeralPlugins;
+
+ public StoragePluginRegistryImpl(DrillbitContext context) {
+ this.context = checkNotNull(context);
+ this.lpPersistence = checkNotNull(context.getLpPersistence());
+ this.classpathScan = checkNotNull(context.getClasspathScan());
+ try {
+ this.pluginSystemTable = context //
+ .getPersistentStoreProvider() //
+ .getStore(PStoreConfig //
+ .newJacksonBuilder(lpPersistence.getMapper(), StoragePluginConfig.class) //
+ .name(PSTORE_NAME) //
+ .build());
+ } catch (IOException | RuntimeException e) {
+ logger.error("Failure while loading storage plugin registry.", e);
+ throw new RuntimeException("Failure while reading and loading storage plugin configuration.", e);
+ }
+
+ ephemeralPlugins = CacheBuilder.newBuilder()
+ .expireAfterAccess(24, TimeUnit.HOURS)
+ .maximumSize(250)
+ .removalListener(new RemovalListener<StoragePluginConfig, StoragePlugin>() {
+ @Override
+ public void onRemoval(RemovalNotification<StoragePluginConfig, StoragePlugin> notification) {
+ closePlugin(notification.getValue());
+ }
+ })
+ .build(new CacheLoader<StoragePluginConfig, StoragePlugin>() {
+ @Override
+ public StoragePlugin load(StoragePluginConfig config) throws Exception {
+ return create(null, config);
+ }
+ });
+ }
+
+ public PStore<StoragePluginConfig> getStore() {
+ return pluginSystemTable;
+ }
+
+ public void init() throws DrillbitStartupException {
+ availablePlugins = findAvailablePlugins(classpathScan);
+
+ // create registered plugins defined in "storage-plugins.json"
+ this.plugins.putAll(createPlugins());
+ }
+
+ private Map<String, StoragePlugin> createPlugins() throws DrillbitStartupException {
+ try {
+ /*
+ * Check if the storage plugins system table has any entries. If not, load the boostrap-storage-plugin file into
+ * the system table.
+ */
+ if (!pluginSystemTable.iterator().hasNext()) {
+ // bootstrap load the config since no plugins are stored.
+ logger.info("No storage plugin instances configured in persistent store, loading bootstrap configuration.");
+ Collection<URL> urls = ClassPathScanner.forResource(ExecConstants.BOOTSTRAP_STORAGE_PLUGINS_FILE, false);
+ if (urls != null && !urls.isEmpty()) {
+ logger.info("Loading the storage plugin configs from URLs {}.", urls);
+ Map<String, URL> pluginURLMap = Maps.newHashMap();
+ for (URL url : urls) {
+ String pluginsData = Resources.toString(url, Charsets.UTF_8);
+ StoragePlugins plugins = lpPersistence.getMapper().readValue(pluginsData, StoragePlugins.class);
+ for (Map.Entry<String, StoragePluginConfig> config : plugins) {
+ if (!pluginSystemTable.putIfAbsent(config.getKey(), config.getValue())) {
+ logger.warn("Duplicate plugin instance '{}' defined in [{}, {}], ignoring the later one.",
+ config.getKey(), pluginURLMap.get(config.getKey()), url);
+ continue;
+ }
+ pluginURLMap.put(config.getKey(), url);
+ }
+ }
+ } else {
+ throw new IOException("Failure finding " + ExecConstants.BOOTSTRAP_STORAGE_PLUGINS_FILE);
+ }
+ }
+
+ Map<String, StoragePlugin> activePlugins = new HashMap<String, StoragePlugin>();
+ for (Map.Entry<String, StoragePluginConfig> entry : pluginSystemTable) {
+ String name = entry.getKey();
+ StoragePluginConfig config = entry.getValue();
+ if (config.isEnabled()) {
+ try {
+ StoragePlugin plugin = create(name, config);
+ activePlugins.put(name, plugin);
+ } catch (ExecutionSetupException e) {
+ logger.error("Failure while setting up StoragePlugin with name: '{}', disabling.", name, e);
+ config.setEnabled(false);
+ pluginSystemTable.put(name, config);
+ }
+ }
+ }
+
+ activePlugins.put(INFORMATION_SCHEMA_PLUGIN, new InfoSchemaStoragePlugin(new InfoSchemaConfig(), context,
+ INFORMATION_SCHEMA_PLUGIN));
+ activePlugins.put(SYS_PLUGIN, new SystemTablePlugin(SystemTablePluginConfig.INSTANCE, context, SYS_PLUGIN));
+
+ return activePlugins;
+ } catch (IOException e) {
+ logger.error("Failure setting up storage plugins. Drillbit exiting.", e);
+ throw new IllegalStateException(e);
+ }
+ }
+
+ @Override
+ public void addPlugin(String name, StoragePlugin plugin) {
+ plugins.put(name, plugin);
+ }
+
+ public void deletePlugin(String name) {
+ StoragePlugin plugin = plugins.remove(name);
+ closePlugin(plugin);
+ pluginSystemTable.delete(name);
+ }
+
+ private void closePlugin(StoragePlugin plugin) {
+ if (plugin == null) {
+ return;
+ }
+
+ try {
+ plugin.close();
+ } catch (Exception e) {
+ logger.warn("Exception while shutting down storage plugin.");
+ }
+ }
+
+ public StoragePlugin createOrUpdate(String name, StoragePluginConfig config, boolean persist)
+ throws ExecutionSetupException {
+ for (;;) {
+ final StoragePlugin oldPlugin = plugins.get(name);
+ final StoragePlugin newPlugin = create(name, config);
+ boolean done = false;
+ try {
+ if (oldPlugin != null) {
+ if (config.isEnabled()) {
+ done = plugins.replace(name, oldPlugin, newPlugin);
+ } else {
+ done = plugins.remove(name, oldPlugin);
+ }
+ if (done) {
+ closePlugin(oldPlugin);
+ }
+ } else if (config.isEnabled()) {
+ done = (null == plugins.putIfAbsent(name, newPlugin));
+ } else {
+ done = true;
+ }
+ } finally {
+ if (!done) {
+ closePlugin(newPlugin);
+ }
+ }
+
+ if (done) {
+ if (persist) {
+ pluginSystemTable.put(name, config);
+ }
+
+ return newPlugin;
+ }
+ }
+ }
+
+ public StoragePlugin getPlugin(String name) throws ExecutionSetupException {
+ StoragePlugin plugin = plugins.get(name);
+ if (name.equals(SYS_PLUGIN) || name.equals(INFORMATION_SCHEMA_PLUGIN)) {
+ return plugin;
+ }
+
+ // since we lazily manage the list of plugins per server, we need to update this once we know that it is time.
+ StoragePluginConfig config = this.pluginSystemTable.get(name);
+ if (config == null) {
+ if (plugin != null) {
+ plugins.remove(name);
+ }
+ return null;
+ } else {
+ if (plugin == null
+ || !plugin.getConfig().equals(config)
+ || plugin.getConfig().isEnabled() != config.isEnabled()) {
+ plugin = createOrUpdate(name, config, false);
+ }
+ return plugin;
+ }
+ }
+
+
+ public StoragePlugin getPlugin(StoragePluginConfig config) throws ExecutionSetupException {
+ if (config instanceof NamedStoragePluginConfig) {
+ return getPlugin(((NamedStoragePluginConfig) config).name);
+ } else {
+ // try to lookup plugin by configuration
+ StoragePlugin plugin = plugins.get(config);
+ if (plugin != null) {
+ return plugin;
+ }
+
+ // no named plugin matches the desired configuration, let's create an
+ // ephemeral storage plugin (or get one from the cache)
+ try {
+ return ephemeralPlugins.get(config);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof ExecutionSetupException) {
+ throw (ExecutionSetupException) cause;
+ } else {
+ // this shouldn't happen. here for completeness.
+ throw new ExecutionSetupException("Failure while trying to create ephemeral plugin.", cause);
+ }
+ }
+ }
+ }
+
+ public FormatPlugin getFormatPlugin(StoragePluginConfig storageConfig, FormatPluginConfig formatConfig)
+ throws ExecutionSetupException {
+ StoragePlugin p = getPlugin(storageConfig);
+ if (!(p instanceof FileSystemPlugin)) {
+ throw new ExecutionSetupException(
+ String.format("You tried to request a format plugin for a storage plugin that wasn't of type "
+ + "FileSystemPlugin. The actual type of plugin was %s.", p.getClass().getName()));
+ }
+ FileSystemPlugin storage = (FileSystemPlugin) p;
+ return storage.getFormatPlugin(formatConfig);
+ }
+
+ private StoragePlugin create(String name, StoragePluginConfig pluginConfig) throws ExecutionSetupException {
+ StoragePlugin plugin = null;
+ Constructor<? extends StoragePlugin> c = availablePlugins.get(pluginConfig.getClass());
+ if (c == null) {
+ throw new ExecutionSetupException(String.format("Failure finding StoragePlugin constructor for config %s",
+ pluginConfig));
+ }
+ try {
+ plugin = c.newInstance(pluginConfig, context, name);
+ plugin.start();
+ return plugin;
+ } catch (InstantiationException | IllegalAccessException | IllegalArgumentException | InvocationTargetException
+ | IOException e) {
+ Throwable t = e instanceof InvocationTargetException ? ((InvocationTargetException) e).getTargetException() : e;
+ if (t instanceof ExecutionSetupException) {
+ throw ((ExecutionSetupException) t);
+ }
+ throw new ExecutionSetupException(String.format(
+ "Failure setting up new storage plugin configuration for config %s", pluginConfig), t);
+ }
+ }
+
+ @Override
+ public Iterator<Entry<String, StoragePlugin>> iterator() {
+ return plugins.iterator();
+ }
+
+ /**
+ * Return StoragePlugin rule sets.
+ *
+ * @param optimizerRulesContext
+ * @return Array of logical and physical rule sets.
+ */
+ public RuleSet[] getStoragePluginRuleSet(OptimizerRulesContext optimizerRulesContext) {
+ // query registered engines for optimizer rules and build the storage plugin RuleSet
+ Builder<RelOptRule> logicalRulesBuilder = ImmutableSet.builder();
+ Builder<RelOptRule> physicalRulesBuilder = ImmutableSet.builder();
+ for (StoragePlugin plugin : this.plugins.plugins()) {
+ Set<? extends RelOptRule> rules = plugin.getLogicalOptimizerRules(optimizerRulesContext);
+ if (rules != null && rules.size() > 0) {
+ logicalRulesBuilder.addAll(rules);
+ }
+ rules = plugin.getPhysicalOptimizerRules(optimizerRulesContext);
+ if (rules != null && rules.size() > 0) {
+ physicalRulesBuilder.addAll(rules);
+ }
+ }
+
+ return new RuleSet[] {
+ DrillRuleSets.create(logicalRulesBuilder.build()),
+ DrillRuleSets.create(physicalRulesBuilder.build()) };
+ }
+
+ public SchemaFactory getSchemaFactory() {
+ return schemaFactory;
+ }
+
+ public class DrillSchemaFactory implements SchemaFactory {
+
+ @Override
+ public void registerSchemas(SchemaConfig schemaConfig, SchemaPlus parent) throws IOException {
+ Stopwatch watch = new Stopwatch();
+ watch.start();
+
+ try {
+ Set<String> currentPluginNames = Sets.newHashSet(plugins.names());
+ // iterate through the plugin instances in the persistence store adding
+ // any new ones and refreshing those whose configuration has changed
+ for (Map.Entry<String, StoragePluginConfig> config : pluginSystemTable) {
+ if (config.getValue().isEnabled()) {
+ getPlugin(config.getKey());
+ currentPluginNames.remove(config.getKey());
+ }
+ }
+ // remove those which are no longer in the registry
+ for (String pluginName : currentPluginNames) {
+ if (pluginName.equals(SYS_PLUGIN) || pluginName.equals(INFORMATION_SCHEMA_PLUGIN)) {
+ continue;
+ }
+ plugins.remove(pluginName);
+ }
+
+ // finally register schemas with the refreshed plugins
+ for (StoragePlugin plugin : plugins.plugins()) {
+ plugin.registerSchemas(schemaConfig, parent);
+ }
+ } catch (ExecutionSetupException e) {
+ throw new DrillRuntimeException("Failure while updating storage plugins", e);
+ }
+
+ // Add second level schema as top level schema with name qualified with parent schema name
+ // Ex: "dfs" schema has "default" and "tmp" as sub schemas. Add following extra schemas "dfs.default" and
+ // "dfs.tmp" under root schema.
+ //
+ // Before change, schema tree looks like below:
+ // "root"
+ // -- "dfs"
+ // -- "default"
+ // -- "tmp"
+ // -- "hive"
+ // -- "default"
+ // -- "hivedb1"
+ //
+ // After the change, the schema tree looks like below:
+ // "root"
+ // -- "dfs"
+ // -- "default"
+ // -- "tmp"
+ // -- "dfs.default"
+ // -- "dfs.tmp"
+ // -- "hive"
+ // -- "default"
+ // -- "hivedb1"
+ // -- "hive.default"
+ // -- "hive.hivedb1"
+ List<SchemaPlus> secondLevelSchemas = Lists.newArrayList();
+ for (String firstLevelSchemaName : parent.getSubSchemaNames()) {
+ SchemaPlus firstLevelSchema = parent.getSubSchema(firstLevelSchemaName);
+ for (String secondLevelSchemaName : firstLevelSchema.getSubSchemaNames()) {
+ secondLevelSchemas.add(firstLevelSchema.getSubSchema(secondLevelSchemaName));
+ }
+ }
+
+ for (SchemaPlus schema : secondLevelSchemas) {
+ AbstractSchema drillSchema;
+ try {
+ drillSchema = schema.unwrap(AbstractSchema.class);
+ } catch (ClassCastException e) {
+ throw new RuntimeException(String.format("Schema '%s' is not expected under root schema", schema.getName()));
+ }
+ SubSchemaWrapper wrapper = new SubSchemaWrapper(drillSchema);
+ parent.add(wrapper.getName(), wrapper);
+ }
+
+ logger.debug("Took {} ms to register schemas.", watch.elapsed(TimeUnit.MILLISECONDS));
+ }
+
+ }
+
+ @Override
+ public synchronized void close() throws Exception {
+ ephemeralPlugins.invalidateAll();
+ plugins.close();
+ }
+
+ /**
+ * Get a list of all available storage plugin class constructors.
+ * @param classpathScan
+ * A classpath scan to use.
+ * @return A Map of StoragePluginConfig => StoragePlugin.<init>() constructors.
+ */
+ @SuppressWarnings("unchecked")
+ public static Map<Object, Constructor<? extends StoragePlugin>> findAvailablePlugins(final ScanResult classpathScan) {
+ Map<Object, Constructor<? extends StoragePlugin>> availablePlugins = new HashMap<Object, Constructor<? extends StoragePlugin>>();
+ final Collection<Class<? extends StoragePlugin>> pluginClasses =
+ classpathScan.getImplementations(StoragePlugin.class);
+ final String lineBrokenList =
+ pluginClasses.size() == 0
+ ? "" : "\n\t- " + Joiner.on("\n\t- ").join(pluginClasses);
+ logger.debug("Found {} storage plugin configuration classes: {}.",
+ pluginClasses.size(), lineBrokenList);
+ for (Class<? extends StoragePlugin> plugin : pluginClasses) {
+ int i = 0;
+ for (Constructor<?> c : plugin.getConstructors()) {
+ Class<?>[] params = c.getParameterTypes();
+ if (params.length != 3
+ || params[1] != DrillbitContext.class
+ || !StoragePluginConfig.class.isAssignableFrom(params[0])
+ || params[2] != String.class) {
+ logger.info("Skipping StoragePlugin constructor {} for plugin class {} since it doesn't implement a "
+ + "[constructor(StoragePluginConfig, DrillbitContext, String)]", c, plugin);
+ continue;
+ }
+ availablePlugins.put(params[0], (Constructor<? extends StoragePlugin>) c);
+ i++;
+ }
+ if (i == 0) {
+ logger.debug("Skipping registration of StoragePlugin {} as it doesn't have a constructor with the parameters "
+ + "of (StorangePluginConfig, Config)", plugin.getCanonicalName());
+ }
+ }
+ return availablePlugins;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/f964908a/exec/java-exec/src/main/resources/drill-module.conf
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/resources/drill-module.conf b/exec/java-exec/src/main/resources/drill-module.conf
index 6b5d9fe..6901a5f 100644
--- a/exec/java-exec/src/main/resources/drill-module.conf
+++ b/exec/java-exec/src/main/resources/drill-module.conf
@@ -75,6 +75,7 @@ drill.exec: {
implementation: "org.apache.drill.exec.opt.IdentityOptimizer"
},
storage: {
+ registry: "org.apache.drill.exec.store.StoragePluginRegistryImpl",
file: {
text: {
buffer.size: 262144,
http://git-wip-us.apache.org/repos/asf/drill/blob/f964908a/exec/java-exec/src/test/java/org/apache/drill/PlanningBase.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/PlanningBase.java b/exec/java-exec/src/test/java/org/apache/drill/PlanningBase.java
index 79245b2..24c8c63 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/PlanningBase.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/PlanningBase.java
@@ -47,6 +47,7 @@ import org.apache.drill.exec.server.options.SessionOptionManager;
import org.apache.drill.exec.server.options.SystemOptionManager;
import org.apache.drill.exec.store.SchemaConfig;
import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.StoragePluginRegistryImpl;
import org.apache.drill.exec.store.sys.local.LocalPStoreProvider;
import org.apache.drill.exec.testing.ExecutionControls;
import org.junit.Rule;
@@ -105,7 +106,7 @@ public class PlanningBase extends ExecTest{
}
};
- final StoragePluginRegistry registry = new StoragePluginRegistry(dbContext);
+ final StoragePluginRegistry registry = new StoragePluginRegistryImpl(dbContext);
registry.init();
final FunctionImplementationRegistry functionRegistry = new FunctionImplementationRegistry(config);
final DrillOperatorTable table = new DrillOperatorTable(functionRegistry);
http://git-wip-us.apache.org/repos/asf/drill/blob/f964908a/exec/java-exec/src/test/java/org/apache/drill/exec/memory/TestAllocators.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/memory/TestAllocators.java b/exec/java-exec/src/test/java/org/apache/drill/exec/memory/TestAllocators.java
index 322e54a..288e78d 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/memory/TestAllocators.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/memory/TestAllocators.java
@@ -47,6 +47,7 @@ import org.apache.drill.exec.server.Drillbit;
import org.apache.drill.exec.server.DrillbitContext;
import org.apache.drill.exec.server.RemoteServiceSet;
import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.StoragePluginRegistryImpl;
import org.apache.drill.exec.vector.BitVector;
import org.apache.drill.exec.vector.IntVector;
import org.apache.drill.test.DrillTest;
@@ -182,7 +183,7 @@ public class TestAllocators extends DrillTest {
bit.run();
final DrillbitContext bitContext = bit.getContext();
FunctionImplementationRegistry functionRegistry = bitContext.getFunctionImplementationRegistry();
- StoragePluginRegistry storageRegistry = new StoragePluginRegistry(bitContext);
+ StoragePluginRegistry storageRegistry = new StoragePluginRegistryImpl(bitContext);
// Create a few Fragment Contexts
http://git-wip-us.apache.org/repos/asf/drill/blob/f964908a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestOptiqPlans.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestOptiqPlans.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestOptiqPlans.java
index 7207bf2..ef102b6 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestOptiqPlans.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestOptiqPlans.java
@@ -54,6 +54,7 @@ import org.apache.drill.exec.server.Drillbit;
import org.apache.drill.exec.server.DrillbitContext;
import org.apache.drill.exec.server.RemoteServiceSet;
import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.StoragePluginRegistryImpl;
import org.apache.drill.exec.store.sys.local.LocalPStoreProvider;
import org.apache.drill.exec.vector.ValueVector;
import org.apache.drill.exec.vector.VarBinaryVector;
@@ -314,7 +315,7 @@ public class TestOptiqPlans extends ExecTest {
}
};
- final StoragePluginRegistry reg = new StoragePluginRegistry(bitContext);
+ final StoragePluginRegistry reg = new StoragePluginRegistryImpl(bitContext);
final PhysicalPlanReader reader = PhysicalPlanReaderTestFactory.defaultPhysicalPlanReader(config, reg);
final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile(file), Charsets.UTF_8));
http://git-wip-us.apache.org/repos/asf/drill/blob/f964908a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java
index 3b363bd..a533620 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java
@@ -23,6 +23,9 @@ import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
+import mockit.Injectable;
+import mockit.NonStrictExpectations;
+
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.scanner.ClassPathScanner;
import org.apache.drill.common.util.FileUtils;
@@ -46,7 +49,7 @@ import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
import org.apache.drill.exec.server.Drillbit;
import org.apache.drill.exec.server.DrillbitContext;
import org.apache.drill.exec.server.RemoteServiceSet;
-import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.StoragePluginRegistryImpl;
import org.apache.drill.exec.vector.ValueVector;
import org.junit.Ignore;
import org.junit.Test;
@@ -56,9 +59,6 @@ import com.google.common.base.Charsets;
import com.google.common.collect.Lists;
import com.google.common.io.Files;
-import mockit.Injectable;
-import mockit.NonStrictExpectations;
-
public class TestMergeJoin extends PopUnitTestBase {
//private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestMergeJoin.class);
@@ -130,7 +130,8 @@ public class TestMergeJoin extends PopUnitTestBase {
bitContext.getCompiler(); result = CodeCompilerTestFactory.getTestCompiler(c);
}};
- final PhysicalPlanReader reader = PhysicalPlanReaderTestFactory.defaultPhysicalPlanReader(c, new StoragePluginRegistry(bitContext));
+ final PhysicalPlanReader reader = PhysicalPlanReaderTestFactory.defaultPhysicalPlanReader(c,
+ new StoragePluginRegistryImpl(bitContext));
final PhysicalPlan plan = reader.readPhysicalPlan(
Files.toString(
FileUtils.getResourceAsFile("/join/merge_single_batch.json"), Charsets.UTF_8)
@@ -186,7 +187,8 @@ public class TestMergeJoin extends PopUnitTestBase {
bitContext.getCompiler(); result = CodeCompilerTestFactory.getTestCompiler(c);
}};
- final PhysicalPlanReader reader = PhysicalPlanReaderTestFactory.defaultPhysicalPlanReader(c, new StoragePluginRegistry(bitContext));
+ final PhysicalPlanReader reader = PhysicalPlanReaderTestFactory.defaultPhysicalPlanReader(c,
+ new StoragePluginRegistryImpl(bitContext));
final PhysicalPlan plan = reader.readPhysicalPlan(
Files.toString(
FileUtils.getResourceAsFile("/join/merge_inner_single_batch.json"), Charsets.UTF_8)
@@ -242,7 +244,8 @@ public class TestMergeJoin extends PopUnitTestBase {
bitContext.getCompiler(); result = CodeCompilerTestFactory.getTestCompiler(c);
}};
- final PhysicalPlanReader reader = PhysicalPlanReaderTestFactory.defaultPhysicalPlanReader(c, new StoragePluginRegistry(bitContext));
+ final PhysicalPlanReader reader = PhysicalPlanReaderTestFactory.defaultPhysicalPlanReader(c,
+ new StoragePluginRegistryImpl(bitContext));
final PhysicalPlan plan = reader.readPhysicalPlan(
Files.toString(
FileUtils.getResourceAsFile("/join/merge_multi_batch.json"), Charsets.UTF_8)