You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@drill.apache.org by ja...@apache.org on 2014/05/07 03:21:00 UTC
[6/6] git commit: DRILL-381: Implement SYSTEM and SESSION options.
DRILL-381: Implement SYSTEM and SESSION options.
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/08923cb8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/08923cb8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/08923cb8
Branch: refs/heads/master
Commit: 08923cb82a007171e699f5ac0b4e4380b3a5a448
Parents: 9b34f23
Author: Jason Altekruse <al...@gmail.com>
Authored: Sat Mar 29 15:02:58 2014 -0500
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Tue May 6 18:15:16 2014 -0700
----------------------------------------------------------------------
.../drill/common/logical/PlanProperties.java | 24 +-
.../drill/exec/cache/DistributedCache.java | 9 +-
.../apache/drill/exec/cache/DistributedMap.java | 3 +-
.../exec/cache/DistributedMapDeserializer.java | 28 +++
.../org/apache/drill/exec/cache/HazelCache.java | 55 ++++-
.../exec/cache/JacksonAdvancedSerializer.java | 65 +++++
.../exec/cache/JacksonDrillSerializable.java | 17 --
.../drill/exec/cache/JacksonSerializable.java | 54 +++++
.../org/apache/drill/exec/cache/LocalCache.java | 94 ++++++-
.../exec/cache/SerializationDefinition.java | 36 +++
.../apache/drill/exec/ops/FragmentContext.java | 22 +-
.../org/apache/drill/exec/ops/QueryContext.java | 9 +-
.../drill/exec/ops/QueryContext.java.orig | 135 +++++++++++
.../apache/drill/exec/opt/BasicOptimizer.java | 63 ++++-
.../org/apache/drill/exec/opt/Optimizer.java | 12 +-
.../drill/exec/planner/PhysicalPlanReader.java | 14 +-
.../drill/exec/planner/SimpleExecPlanner.java | 15 +-
.../planner/fragment/SimpleParallelizer.java | 11 +-
.../exec/planner/logical/StoragePlugins.java | 13 +-
.../exec/planner/physical/PlannerSettings.java | 17 +-
.../planner/sql/handlers/DefaultSqlHandler.java | 4 +-
.../planner/sql/handlers/ExplainHandler.java | 12 +-
.../planner/sql/handlers/SetOptionHandler.java | 20 +-
.../sql/handlers/SimpleCommandResult.java | 1 -
.../apache/drill/exec/rpc/user/UserServer.java | 10 +-
.../apache/drill/exec/rpc/user/UserSession.java | 16 +-
.../org/apache/drill/exec/server/Drillbit.java | 2 +
.../apache/drill/exec/server/Drillbit.java.orig | 141 +++++++++++
.../drill/exec/server/DrillbitContext.java | 12 +-
.../drill/exec/server/DrillbitContext.java.orig | 157 ++++++++++++
.../server/options/DrillConfigIterator.java | 85 +++++++
.../server/options/FragmentOptionsManager.java | 81 +++++++
.../drill/exec/server/options/OptionList.java | 23 ++
.../exec/server/options/OptionManager.java | 37 +++
.../exec/server/options/OptionValidator.java | 65 +++++
.../drill/exec/server/options/OptionValue.java | 130 ++++++++++
.../server/options/SessionOptionManager.java | 98 ++++++++
.../exec/server/options/SetOptionException.java | 62 +++++
.../server/options/SystemOptionManager.java | 154 ++++++++++++
.../exec/server/options/TypeValidators.java | 132 ++++++++++
.../drill/exec/store/StoragePluginRegistry.java | 34 +--
.../exec/store/StoragePluginRegistry.java.orig | 243 +++++++++++++++++++
.../exec/store/direct/DirectGroupScan.java | 7 +
.../drill/exec/store/mock/MockGroupScanPOP.java | 27 ++-
.../drill/exec/store/pojo/AbstractWriter.java | 2 +-
.../drill/exec/store/pojo/PojoDataType.java | 72 ++++++
.../drill/exec/store/pojo/PojoRecordReader.java | 20 +-
.../apache/drill/exec/store/pojo/Writers.java | 133 +++++++++-
.../drill/exec/store/sys/DrillbitIterator.java | 61 +++++
.../drill/exec/store/sys/StaticDrillTable.java | 41 ++++
.../drill/exec/store/sys/SystemTable.java | 55 +++++
.../exec/store/sys/SystemTableBatchCreator.java | 44 ++++
.../drill/exec/store/sys/SystemTablePlugin.java | 118 +++++++++
.../exec/store/sys/SystemTablePluginConfig.java | 32 +++
.../drill/exec/store/sys/SystemTableScan.java | 128 ++++++++++
.../exec/work/batch/ControlHandlerImpl.java | 40 +--
.../exec/work/batch/ControlMessageHandler.java | 4 +-
.../apache/drill/exec/work/foreman/Foreman.java | 12 +-
.../work/fragment/NonRootFragmentManager.java | 6 +-
.../apache/drill/exec/work/user/UserWorker.java | 5 +
.../java/org/apache/drill/PlanningBase.java | 18 +-
.../org/apache/drill/TestTpchSingleMode.java | 3 +-
.../exec/physical/impl/TestOptiqPlans.java | 6 +-
.../drill/exec/pop/TestFragmentChecker.java | 29 +--
.../apache/drill/exec/server/TestOptions.java | 41 ++++
.../resources/server/options_session_check.json | 20 ++
.../src/test/resources/server/options_set.json | 24 ++
.../apache/drill/jdbc/test/TestJdbcQuery.java | 3 +
.../org/apache/drill/exec/proto/BitControl.java | 181 +++++++++++++-
protocol/src/main/protobuf/BitControl.proto | 1 +
70 files changed, 3142 insertions(+), 206 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/common/src/main/java/org/apache/drill/common/logical/PlanProperties.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/drill/common/logical/PlanProperties.java b/common/src/main/java/org/apache/drill/common/logical/PlanProperties.java
index 606f2df..a2942b6 100644
--- a/common/src/main/java/org/apache/drill/common/logical/PlanProperties.java
+++ b/common/src/main/java/org/apache/drill/common/logical/PlanProperties.java
@@ -6,9 +6,9 @@
* 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.
@@ -17,10 +17,9 @@
******************************************************************************/
package org.apache.drill.common.logical;
+import org.apache.drill.common.JSONOptions;
import org.apache.drill.common.logical.PlanProperties.Generator.ResultMode;
-import com.fasterxml.jackson.annotation.JsonInclude;
-import com.fasterxml.jackson.annotation.JsonInclude.Include;
import com.fasterxml.jackson.annotation.JsonProperty;
/**
@@ -33,12 +32,13 @@ public class PlanProperties {
public int version;
public Generator generator;
public ResultMode resultMode;
+ public JSONOptions options;
// @JsonInclude(Include.NON_NULL)
public static class Generator{
public String type;
public String info;
-
+
public static enum ResultMode{
EXEC, LOGICAL, PHYSICAL;
}
@@ -52,12 +52,14 @@ public class PlanProperties {
private PlanProperties(@JsonProperty("version") int version,
@JsonProperty("generator") Generator generator,
@JsonProperty("type") PlanType type,
- @JsonProperty("mode") ResultMode resultMode
+ @JsonProperty("mode") ResultMode resultMode,
+ @JsonProperty("options") JSONOptions options
) {
this.version = version;
this.generator = generator;
this.type = type;
this.resultMode = resultMode == null ? ResultMode.EXEC : resultMode;
+ this.options = options;
}
public static PlanPropertiesBuilder builder() {
@@ -69,6 +71,7 @@ public class PlanProperties {
private Generator generator;
private PlanType type;
private ResultMode mode = ResultMode.EXEC;
+ private JSONOptions options;
public PlanPropertiesBuilder type(PlanType type) {
this.type = type;
@@ -84,19 +87,24 @@ public class PlanProperties {
this.generator = new Generator(type, info);
return this;
}
-
+
public PlanPropertiesBuilder resultMode(ResultMode mode){
this.mode = mode;
return this;
}
+ public PlanPropertiesBuilder options(JSONOptions options){
+ this.options = options;
+ return this;
+ }
+
public PlanPropertiesBuilder generator(Generator generator) {
this.generator = generator;
return this;
}
public PlanProperties build() {
- return new PlanProperties(version, generator, type, mode);
+ return new PlanProperties(version, generator, type, mode, options);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedCache.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedCache.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedCache.java
index fdace08..65362e0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedCache.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedCache.java
@@ -26,15 +26,16 @@ import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
public interface DistributedCache extends Closeable{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DistributedCache.class);
-
+
public void run() throws DrillbitStartupException;
-
+
// public void updateLocalQueueLength(int length);
-// public List<WorkQueueStatus> getQueueLengths();
-
+// public List<WorkQueueStatus> getQueueLengths();
+
public PlanFragment getFragment(FragmentHandle handle);
public void storeFragment(PlanFragment fragment);
public <V extends DrillSerializable> DistributedMultiMap<V> getMultiMap(Class<V> clazz);
public <V extends DrillSerializable> DistributedMap<V> getMap(Class<V> clazz);
+ public <V extends DrillSerializable> DistributedMap<V> getNamedMap(String name, Class<V> clazz);
public Counter getCounter(String name);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedMap.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedMap.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedMap.java
index 8ea9cd1..1bee5fc 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedMap.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedMap.java
@@ -17,9 +17,10 @@
*/
package org.apache.drill.exec.cache;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
-public interface DistributedMap<V extends DrillSerializable> {
+public interface DistributedMap<V extends DrillSerializable> extends Iterable<Map.Entry<String, V>>{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DistributedMap.class);
public V get(String key);
public void put(String key, V value);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedMapDeserializer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedMapDeserializer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedMapDeserializer.java
new file mode 100644
index 0000000..453dbfb
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/DistributedMapDeserializer.java
@@ -0,0 +1,28 @@
+/**
+ * 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.cache;
+
+import java.io.IOException;
+
+public interface DistributedMapDeserializer<V extends DrillSerializable> {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DistributedMapDeserializer.class);
+
+ public V put(String key, byte[] value) throws IOException;
+
+ public Class getValueClass();
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/cache/HazelCache.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/HazelCache.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/HazelCache.java
index b701c8f..0149a57 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/HazelCache.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/HazelCache.java
@@ -19,7 +19,9 @@ package org.apache.drill.exec.cache;
import java.io.IOException;
import java.util.Collection;
+import java.util.Iterator;
import java.util.List;
+import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import org.apache.drill.common.config.DrillConfig;
@@ -45,6 +47,7 @@ import com.hazelcast.core.IMap;
import com.hazelcast.core.ITopic;
import com.hazelcast.core.Message;
import com.hazelcast.core.MessageListener;
+import com.hazelcast.nio.serialization.StreamSerializer;
public class HazelCache implements DistributedCache {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(HazelCache.class);
@@ -55,12 +58,26 @@ public class HazelCache implements DistributedCache {
private HandlePlan fragments;
private Cache<WorkQueueStatus, Integer> endpoints;
private BufferAllocator allocator;
+ private DrillConfig config;
public HazelCache(DrillConfig config, BufferAllocator allocator) {
this.instanceName = config.getString(ExecConstants.SERVICE_NAME);
this.allocator = allocator;
+ this.config = config;
}
+ private <T> void addSer(Config c, StreamSerializer<T> serializer, Class<T> clazz){
+ SerializerConfig sc = new SerializerConfig().setImplementation(serializer).setTypeClass(clazz);
+ c.getSerializationConfig().addSerializerConfig(sc);
+ }
+
+ @SuppressWarnings("rawtypes")
+ private <T> void addJSer(Config c, SerializationDefinition d){
+ SerializerConfig sc = new SerializerConfig().setImplementation(new JacksonAdvancedSerializer(d, config.getMapper())).setTypeClass(d.clazz);
+ c.getSerializationConfig().addSerializerConfig(sc);
+ }
+
+
private class Listener implements MessageListener<HWorkQueueStatus>{
@Override
@@ -68,16 +85,16 @@ public class HazelCache implements DistributedCache {
logger.debug("Received new queue length message.");
endpoints.put(wrapped.getMessageObject().get(), 0);
}
-
+
}
-
+
public void run() {
Config c = new Config();
- SerializerConfig sc = new SerializerConfig() //
- .setImplementation(new HCVectorAccessibleSerializer(allocator)) //
- .setTypeClass(VectorAccessibleSerializable.class);
+ addSer(c, new HCVectorAccessibleSerializer(allocator), VectorAccessibleSerializable.class);
+ addJSer(c, SerializationDefinition.OPTION);
+ addJSer(c, SerializationDefinition.STORAGE_PLUGINS);
+
c.setInstanceName(instanceName);
- c.getSerializationConfig().addSerializerConfig(sc);
c.getGroupConfig().setName(instanceName);
for (String s : DrillConfig.create().getStringList(ExecConstants.HAZELCAST_SUBNETS)) {
logger.debug("Adding interface: {}", s);
@@ -128,7 +145,7 @@ public class HazelCache implements DistributedCache {
public void storeFragment(PlanFragment fragment) {
fragments.put(fragment.getHandle(), fragment);
}
-
+
@Override
public <V extends DrillSerializable> DistributedMultiMap<V> getMultiMap(Class<V> clazz) {
@@ -138,12 +155,18 @@ public class HazelCache implements DistributedCache {
@Override
public <V extends DrillSerializable> DistributedMap<V> getMap(Class<V> clazz) {
- IMap<String, V> imap = this.instance.getMap(clazz.toString());
+ return getNamedMap(clazz.getName(), clazz);
+ }
+
+
+ @Override
+ public <V extends DrillSerializable> DistributedMap<V> getNamedMap(String name, Class<V> clazz) {
+ IMap<String, V> imap = this.instance.getMap(name);
MapConfig myMapConfig = new MapConfig();
myMapConfig.setBackupCount(0);
myMapConfig.setReadBackupData(true);
instance.getConfig().getMapConfigs().put(clazz.toString(), myMapConfig);
- return new HCDistributedMapImpl<V>(imap, clazz);
+ return new HCDistributedMapImpl<V>(imap);
}
@Override
@@ -151,10 +174,12 @@ public class HazelCache implements DistributedCache {
return new HCCounterImpl(this.instance.getAtomicLong(name));
}
+
+
public static class HCDistributedMapImpl<V extends DrillSerializable> implements DistributedMap<V> {
- private IMap<String, V> m;
+ private final IMap<String, V> m;
- public HCDistributedMapImpl(IMap<String, V> m, Class<V> clazz) {
+ public HCDistributedMapImpl(IMap<String, V> m) {
this.m = m;
}
@@ -172,7 +197,15 @@ public class HazelCache implements DistributedCache {
public void putIfAbsent(String key, V value, long ttl, TimeUnit timeunit) {
m.putIfAbsent(key, value, ttl, timeunit);
+
+ }
+
+ @Override
+ public Iterator<Entry<String, V>> iterator() {
+ return m.entrySet().iterator();
}
+
+
}
public static class HCDistributedMultiMapImpl<V extends DrillSerializable> implements DistributedMultiMap<V> {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/cache/JacksonAdvancedSerializer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/JacksonAdvancedSerializer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/JacksonAdvancedSerializer.java
new file mode 100644
index 0000000..edf31aa
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/JacksonAdvancedSerializer.java
@@ -0,0 +1,65 @@
+/**
+ * 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.cache;
+
+import java.io.IOException;
+
+import org.apache.drill.common.util.DataInputInputStream;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.hazelcast.nio.ObjectDataInput;
+import com.hazelcast.nio.ObjectDataOutput;
+import com.hazelcast.nio.serialization.StreamSerializer;
+
+public class JacksonAdvancedSerializer<T> implements StreamSerializer<T> {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(JacksonAdvancedSerializer.class);
+
+ private final Class<?> clazz;
+ private final ObjectMapper mapper;
+ private final int id;
+
+ public JacksonAdvancedSerializer(SerializationDefinition def, ObjectMapper mapper){
+ this.clazz = def.clazz;
+ this.mapper = mapper;
+ this.id = def.id;
+ }
+
+ @Override
+ public int getTypeId() {
+ return id;
+ }
+
+ @Override
+ public void destroy() {
+ }
+
+ @Override
+ public void write(ObjectDataOutput out, T object) throws IOException {
+ out.write(mapper.writeValueAsBytes(object));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public T read(ObjectDataInput in) throws IOException {
+ return (T) mapper.readValue(DataInputInputStream.constructInputStream(in), clazz);
+ }
+
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/cache/JacksonDrillSerializable.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/JacksonDrillSerializable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/JacksonDrillSerializable.java
index a7b0be2..dcfc1ec 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/JacksonDrillSerializable.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/JacksonDrillSerializable.java
@@ -66,21 +66,4 @@ public abstract class JacksonDrillSerializable<T> implements DrillSerializable,
return obj;
}
- public static class StoragePluginsSerializable extends JacksonDrillSerializable<StoragePlugins> {
-
- public StoragePluginsSerializable(DrillbitContext context, StoragePlugins obj) {
- super(context, obj);
- }
-
- public StoragePluginsSerializable(BufferAllocator allocator) {
- }
-
- public StoragePluginsSerializable() {
- }
-
- @Override
- public void readFromStream(InputStream input) throws IOException {
- readFromStream(input, StoragePlugins.class);
- }
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/cache/JacksonSerializable.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/JacksonSerializable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/JacksonSerializable.java
new file mode 100644
index 0000000..831db84
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/JacksonSerializable.java
@@ -0,0 +1,54 @@
+/**
+ * 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.cache;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import com.hazelcast.nio.ObjectDataInput;
+import com.hazelcast.nio.ObjectDataOutput;
+
+public abstract class JacksonSerializable implements DrillSerializable{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(JacksonSerializable.class);
+
+ private void fail(){
+ throw new UnsupportedOperationException("Need to register serialization config for class " + this.getClass().getName()); // rely on external serializer
+
+ }
+
+ @Override
+ public void readData(ObjectDataInput input) throws IOException {
+ fail();
+ }
+
+ @Override
+ public void readFromStream(InputStream input) throws IOException {
+ fail();
+ }
+
+ @Override
+ public void writeData(ObjectDataOutput output) throws IOException {
+ fail();
+ }
+
+ @Override
+ public void writeToStream(OutputStream output) throws IOException {
+ fail();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/cache/LocalCache.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/LocalCache.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/LocalCache.java
index 119764b..0fb4b82 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/LocalCache.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/LocalCache.java
@@ -17,14 +17,15 @@
*/
package org.apache.drill.exec.cache;
-import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.InvocationTargetException;
import java.util.Collection;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -38,6 +39,7 @@ import org.apache.drill.exec.memory.TopLevelAllocator;
import org.apache.drill.exec.proto.BitControl.PlanFragment;
import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -49,11 +51,14 @@ public class LocalCache implements DistributedCache {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(LocalCache.class);
private volatile Map<FragmentHandle, PlanFragment> handles;
+ private volatile ConcurrentMap<String, DistributedMap<?>> namedMaps;
private volatile ConcurrentMap<Class<?>, DistributedMap<?>> maps;
private volatile ConcurrentMap<Class<?>, DistributedMultiMap<?>> multiMaps;
private volatile ConcurrentMap<String, Counter> counters;
private static final BufferAllocator allocator = new TopLevelAllocator();
+ private static final ObjectMapper mapper = DrillConfig.create().getMapper();
+
@Override
public void close() throws IOException {
handles = null;
@@ -65,6 +70,7 @@ public class LocalCache implements DistributedCache {
maps = Maps.newConcurrentMap();
multiMaps = Maps.newConcurrentMap();
counters = Maps.newConcurrentMap();
+ namedMaps = Maps.newConcurrentMap();
}
@Override
@@ -78,7 +84,7 @@ public class LocalCache implements DistributedCache {
// logger.debug("Storing fragment: {}", fragment);
handles.put(fragment.getHandle(), fragment);
}
-
+
@Override
public <V extends DrillSerializable> DistributedMultiMap<V> getMultiMap(Class<V> clazz) {
DistributedMultiMap<V> mmap = (DistributedMultiMap<V>) multiMaps.get(clazz);
@@ -101,6 +107,18 @@ public class LocalCache implements DistributedCache {
}
}
+
+ @Override
+ public <V extends DrillSerializable> DistributedMap<V> getNamedMap(String name, Class<V> clazz) {
+ DistributedMap m = namedMaps.get(clazz);
+ if (m == null) {
+ namedMaps.putIfAbsent(name, new LocalDistributedMapImpl<V>(clazz));
+ return (DistributedMap<V>) namedMaps.get(name);
+ } else {
+ return m;
+ }
+ }
+
@Override
public Counter getCounter(String name) {
Counter c = counters.get(name);
@@ -113,6 +131,16 @@ public class LocalCache implements DistributedCache {
}
public static ByteArrayDataOutput serialize(DrillSerializable obj) {
+ if(obj instanceof JacksonSerializable){
+ try{
+ ByteArrayDataOutput out = ByteStreams.newDataOutput();
+ out.write(mapper.writeValueAsBytes(obj));
+ return out;
+ }catch(Exception e){
+ throw new RuntimeException(e);
+ }
+ }
+
ByteArrayDataOutput out = ByteStreams.newDataOutput();
OutputStream outputStream = DataOutputOutputStream.constructOutputStream(out);
try {
@@ -129,6 +157,14 @@ public class LocalCache implements DistributedCache {
}
public static <V extends DrillSerializable> V deserialize(byte[] bytes, Class<V> clazz) {
+ if(JacksonSerializable.class.isAssignableFrom(clazz)){
+ try{
+ return (V) mapper.readValue(bytes, clazz);
+ }catch(Exception e){
+ throw new RuntimeException(e);
+ }
+ }
+
ByteArrayDataInput in = ByteStreams.newDataInput(bytes);
InputStream inputStream = DataInputInputStream.constructInputStream(in);
try {
@@ -165,8 +201,8 @@ public class LocalCache implements DistributedCache {
}
public static class LocalDistributedMapImpl<V extends DrillSerializable> implements DistributedMap<V> {
- private ConcurrentMap<String, ByteArrayDataOutput> m;
- private Class<V> clazz;
+ protected ConcurrentMap<String, ByteArrayDataOutput> m;
+ protected Class<V> clazz;
public LocalDistributedMapImpl(Class<V> clazz) {
m = Maps.newConcurrentMap();
@@ -196,6 +232,56 @@ public class LocalCache implements DistributedCache {
m.putIfAbsent(key, serialize(value));
logger.warn("Expiration not implemented in local map cache");
}
+
+ private class DeserializingTransformer implements Iterator<Map.Entry<String, V> >{
+ private Iterator<Map.Entry<String, ByteArrayDataOutput>> inner;
+
+ public DeserializingTransformer(Iterator<Entry<String, ByteArrayDataOutput>> inner) {
+ super();
+ this.inner = inner;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return inner.hasNext();
+ }
+
+ @Override
+ public Entry<String, V> next() {
+ return newEntry(inner.next());
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ public Entry<String, V> newEntry(final Entry<String, ByteArrayDataOutput> input) {
+ return new Map.Entry<String, V>(){
+
+ @Override
+ public String getKey() {
+ return input.getKey();
+ }
+
+ @Override
+ public V getValue() {
+ return deserialize(input.getValue().toByteArray(), clazz);
+ }
+
+ @Override
+ public V setValue(V value) {
+ throw new UnsupportedOperationException();
+ }
+
+ };
+ }
+
+ }
+ @Override
+ public Iterator<Entry<String, V>> iterator() {
+ return new DeserializingTransformer(m.entrySet().iterator());
+ }
}
public static class LocalCounterImpl implements Counter {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/cache/SerializationDefinition.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/SerializationDefinition.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/SerializationDefinition.java
new file mode 100644
index 0000000..95ba434
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/SerializationDefinition.java
@@ -0,0 +1,36 @@
+/**
+ * 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.cache;
+
+import org.apache.drill.exec.planner.logical.StoragePlugins;
+import org.apache.drill.exec.server.options.OptionValue;
+
+public enum SerializationDefinition {
+
+ OPTION(3002, OptionValue.class),
+ STORAGE_PLUGINS(3003, StoragePlugins.class)
+ ;
+ public final int id;
+ public final Class<?> clazz;
+
+ SerializationDefinition(int id, Class<?> clazz){
+ this.id = id;
+ this.clazz = clazz;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
index f3bcfef..2a6ab0e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
@@ -27,6 +27,7 @@ import net.hydromatic.optiq.tools.Frameworks;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.exec.compile.ClassTransformer;
import org.apache.drill.exec.compile.QueryClassLoader;
import org.apache.drill.exec.exception.ClassTransformationException;
@@ -42,6 +43,9 @@ import org.apache.drill.exec.rpc.control.ControlTunnel;
import org.apache.drill.exec.rpc.data.DataTunnel;
import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.server.options.FragmentOptionsManager;
+import org.apache.drill.exec.server.options.OptionList;
+import org.apache.drill.exec.server.options.OptionManager;
import org.apache.drill.exec.work.batch.IncomingBuffers;
import com.beust.jcommander.internal.Lists;
@@ -68,12 +72,13 @@ public class FragmentContext implements Closeable {
private IncomingBuffers buffers;
private final long queryStartTime;
private final int rootFragmentTimeZone;
+ private final OptionManager sessionOptions;
private volatile Throwable failureCause;
private volatile boolean failed = false;
public FragmentContext(DrillbitContext dbContext, PlanFragment fragment, UserClientConnection connection,
- FunctionImplementationRegistry funcRegistry) throws OutOfMemoryException {
+ FunctionImplementationRegistry funcRegistry) throws OutOfMemoryException, ExecutionSetupException {
this.loader = new QueryClassLoader(true);
this.transformer = new ClassTransformer();
this.stats = new FragmentStats(dbContext.getMetrics());
@@ -85,9 +90,24 @@ public class FragmentContext implements Closeable {
this.rootFragmentTimeZone = fragment.getTimeZone();
logger.debug("Getting initial memory allocation of {}", fragment.getMemInitial());
logger.debug("Fragment max allocation: {}", fragment.getMemMax());
+ try{
+ OptionList list;
+ if(!fragment.hasOptionsJson() || fragment.getOptionsJson().isEmpty()){
+ list = new OptionList();
+ }else{
+ list = dbContext.getConfig().getMapper().readValue(fragment.getOptionsJson(), OptionList.class);
+ }
+ this.sessionOptions = new FragmentOptionsManager(context.getOptionManager(), list);
+ }catch(Exception e){
+ throw new ExecutionSetupException("Failure while reading plan options.", e);
+ }
this.allocator = context.getAllocator().getChildAllocator(fragment.getHandle(), fragment.getMemInitial(), fragment.getMemMax());
}
+ public OptionManager getOptions(){
+ return sessionOptions;
+ }
+
public void setBuffers(IncomingBuffers buffers) {
this.buffers = buffers;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java
index 7e3b63d..f6ce04f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java
@@ -23,6 +23,7 @@ import net.hydromatic.optiq.SchemaPlus;
import net.hydromatic.optiq.tools.Frameworks;
import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.exceptions.ExpressionParsingException;
import org.apache.drill.exec.cache.DistributedCache;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
import org.apache.drill.exec.planner.PhysicalPlanReader;
@@ -33,7 +34,10 @@ import org.apache.drill.exec.rpc.control.WorkEventBus;
import org.apache.drill.exec.rpc.data.DataConnectionCreator;
import org.apache.drill.exec.rpc.user.UserSession;
import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.exec.server.options.OptionValue;
import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.eigenbase.sql.SqlLiteral;
public class QueryContext{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(QueryContext.class);
@@ -52,7 +56,7 @@ public class QueryContext{
this.workBus = drllbitContext.getWorkBus();
this.session = session;
this.timer = new Multitimer<>(QuerySetup.class);
- this.plannerSettings = new PlannerSettings();
+ this.plannerSettings = new PlannerSettings(session.getOptions());
}
public PlannerSettings getPlannerSettings(){
@@ -79,6 +83,9 @@ public class QueryContext{
return rootSchema;
}
+ public OptionManager getOptions(){
+ return session.getOptions();
+ }
public DrillbitEndpoint getCurrentEndpoint(){
return drillbitContext.getEndpoint();
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java.orig
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java.orig b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java.orig
new file mode 100644
index 0000000..4f050a6
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/QueryContext.java.orig
@@ -0,0 +1,135 @@
+/**
+ * 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.ops;
+
+import java.util.Collection;
+
+import net.hydromatic.optiq.SchemaPlus;
+import net.hydromatic.optiq.tools.Frameworks;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.exceptions.ExpressionParsingException;
+import org.apache.drill.exec.cache.DistributedCache;
+import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.planner.physical.PlannerSettings;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.rpc.control.WorkEventBus;
+import org.apache.drill.exec.rpc.data.DataConnectionCreator;
+import org.apache.drill.exec.rpc.user.UserSession;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+
+public class QueryContext{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(QueryContext.class);
+
+ private final QueryId queryId;
+ private final DrillbitContext drillbitContext;
+ private final WorkEventBus workBus;
+ private UserSession session;
+ public final Multitimer<QuerySetup> timer;
+ private final PlannerSettings plannerSettings;
+
+ public QueryContext(UserSession session, QueryId queryId, DrillbitContext drllbitContext) {
+ super();
+ this.queryId = queryId;
+ this.drillbitContext = drllbitContext;
+ this.workBus = drllbitContext.getWorkBus();
+ this.session = session;
+ this.timer = new Multitimer<>(QuerySetup.class);
+ this.plannerSettings = new PlannerSettings();
+ }
+
+ public PlannerSettings getPlannerSettings(){
+ return plannerSettings;
+ }
+
+ public UserSession getSession(){
+ return session;
+ }
+
+ public SchemaPlus getNewDefaultSchema(){
+ SchemaPlus rootSchema = getRootSchema();
+ SchemaPlus defaultSchema = session.getDefaultSchema(rootSchema);
+ if(defaultSchema == null){
+ return rootSchema;
+ }else{
+ return defaultSchema;
+ }
+ }
+
+ public SchemaPlus getRootSchema(){
+ SchemaPlus rootSchema = Frameworks.createRootSchema();
+ drillbitContext.getSchemaFactory().registerSchemas(session.getUser(), rootSchema);
+ return rootSchema;
+ }
+
+<<<<<<< HEAD
+=======
+ public Object getOptionValue(String name){
+ return drillbitContext.getGlobalDrillOptions().getOption(name);
+ }
+
+ public void setOptionValue(String name, String value) throws ExpressionParsingException {
+ drillbitContext.getGlobalDrillOptions().setOption(name, value);
+ }
+
+>>>>>>> Drill 381 - implementation of session and global options.
+ public DrillbitEndpoint getCurrentEndpoint(){
+ return drillbitContext.getEndpoint();
+ }
+
+ public QueryId getQueryId() {
+ return queryId;
+ }
+
+ public StoragePluginRegistry getStorage(){
+ return drillbitContext.getStorage();
+ }
+
+
+ public DistributedCache getCache(){
+ return drillbitContext.getCache();
+ }
+
+ public Collection<DrillbitEndpoint> getActiveEndpoints(){
+ return drillbitContext.getBits();
+ }
+
+ public PhysicalPlanReader getPlanReader(){
+ return drillbitContext.getPlanReader();
+ }
+
+ public DataConnectionCreator getDataConnectionsPool(){
+ return drillbitContext.getDataConnectionsPool();
+ }
+
+ public DrillConfig getConfig(){
+ return drillbitContext.getConfig();
+ }
+
+ public WorkEventBus getWorkBus(){
+ return workBus;
+ }
+
+ public FunctionImplementationRegistry getFunctionRegistry(){
+ return drillbitContext.getFunctionImplementationRegistry();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/opt/BasicOptimizer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/opt/BasicOptimizer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/opt/BasicOptimizer.java
index d7d5ccb..78c0e3c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/opt/BasicOptimizer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/opt/BasicOptimizer.java
@@ -23,6 +23,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.List;
+import org.apache.drill.common.JSONOptions;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.common.logical.LogicalPlan;
@@ -53,8 +54,8 @@ import org.apache.drill.exec.physical.config.Screen;
import org.apache.drill.exec.physical.config.SelectionVectorRemover;
import org.apache.drill.exec.physical.config.Sort;
import org.apache.drill.exec.physical.config.StreamingAggregate;
-import org.apache.drill.exec.physical.config.HashAggregate;
-import org.apache.drill.exec.physical.OperatorCost;
+import org.apache.drill.exec.rpc.user.UserServer;
+import org.apache.drill.exec.server.options.OptionManager;
import org.apache.drill.exec.store.StoragePlugin;
import org.eigenbase.rel.RelFieldCollation.Direction;
import org.eigenbase.rel.RelFieldCollation.NullDirection;
@@ -62,13 +63,46 @@ import org.eigenbase.rel.RelFieldCollation.NullDirection;
import com.beust.jcommander.internal.Lists;
public class BasicOptimizer extends Optimizer{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BasicOptimizer.class);
private DrillConfig config;
private QueryContext context;
+ private UserServer.UserClientConnection userSession;
- public BasicOptimizer(DrillConfig config, QueryContext context){
+ public BasicOptimizer(DrillConfig config, QueryContext context, UserServer.UserClientConnection userSession){
this.config = config;
this.context = context;
+ this.userSession = userSession;
+ logCurrentOptionValues();
+ }
+
+ private void logCurrentOptionValues(){
+// Iterator<DrillOptionValue> optionVals = userSession.getSessionOptionIterator();
+// DrillOptionValue val = null;
+// String output = "";
+// output += "SessionOptions: {\n";
+// for ( ;optionVals.hasNext(); val = optionVals.next()){
+// if (val != null) {
+// output += val.getOptionName() + ":" + val.getValue() + ",\n";
+// }
+// }
+// output += "}";
+// logger.debug(output);
+ }
+
+ /**
+ * Get the current value of an option. Session options override global options.
+ *
+ * @param name - the name of the option
+ * @return - value of the option
+ */
+ private Object getOptionValue(String name) {
+// Object val = userSession.getSessionLevelOption(name);
+// if (val == null) {
+// context.getOptionValue(name);
+// }
+// return val;
+ return null;
}
@Override
@@ -90,7 +124,8 @@ public class BasicOptimizer extends Optimizer{
PlanProperties props = PlanProperties.builder()
.type(PlanProperties.PlanType.APACHE_DRILL_PHYSICAL)
.version(plan.getProperties().version)
- .generator(plan.getProperties().generator).build();
+ .generator(plan.getProperties().generator)
+ .options(new JSONOptions(context.getOptions().getSessionOptionList())).build();
PhysicalPlan p = new PhysicalPlan(props, physOps);
return p;
//return new PhysicalPlan(props, physOps);
@@ -103,10 +138,20 @@ public class BasicOptimizer extends Optimizer{
public static class BasicOptimizationContext implements OptimizationContext {
+ private OptionManager ops;
+ public BasicOptimizationContext(QueryContext c){
+ this.ops = c.getOptions();
+ }
+
@Override
public int getPriority() {
return 1;
}
+
+ @Override
+ public OptionManager getOptions() {
+ return ops;
+ }
}
private class LogicalConverter extends AbstractLogicalVisitor<PhysicalOperator, Object, OptimizerException> {
@@ -121,10 +166,10 @@ public class BasicOptimizer extends Optimizer{
@Override
public PhysicalOperator visitGroupingAggregate(GroupingAggregate groupBy, Object value) throws OptimizerException {
-
+
List<Ordering> orderDefs = Lists.newArrayList();
-
+
PhysicalOperator input = groupBy.getInput().accept(this, value);
if(groupBy.getKeys().length > 0){
@@ -133,7 +178,7 @@ public class BasicOptimizer extends Optimizer{
}
input = new Sort(input, orderDefs, false);
}
-
+
StreamingAggregate sa = new StreamingAggregate(input, groupBy.getKeys(), groupBy.getExprs(), 1.0f);
return sa;
}
@@ -165,7 +210,7 @@ public class BasicOptimizer extends Optimizer{
}
leftOp = new Sort(leftOp, leftOrderDefs, false);
leftOp = new SelectionVectorRemover(leftOp);
-
+
PhysicalOperator rightOp = join.getRight().accept(this, value);
List<Ordering> rightOrderDefs = Lists.newArrayList();
for(JoinCondition jc : join.getConditions()){
@@ -173,7 +218,7 @@ public class BasicOptimizer extends Optimizer{
}
rightOp = new Sort(rightOp, rightOrderDefs, false);
rightOp = new SelectionVectorRemover(rightOp);
-
+
MergeJoinPOP mjp = new MergeJoinPOP(leftOp, rightOp, Arrays.asList(join.getConditions()), join.getJoinType());
return new SelectionVectorRemover(mjp);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/opt/Optimizer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/opt/Optimizer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/opt/Optimizer.java
index c7069c3..34d0622 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/opt/Optimizer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/opt/Optimizer.java
@@ -24,23 +24,25 @@ import org.apache.drill.common.exceptions.DrillConfigurationException;
import org.apache.drill.common.logical.LogicalPlan;
import org.apache.drill.exec.exception.OptimizerException;
import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.server.options.OptionManager;
public abstract class Optimizer implements Closeable{
-
+
public static String OPTIMIZER_IMPL_KEY = "drill.exec.optimizer.implementation";
-
+
public abstract void init(DrillConfig config);
-
+
public abstract PhysicalPlan optimize(OptimizationContext context, LogicalPlan plan) throws OptimizerException;
public abstract void close();
-
+
public static Optimizer getOptimizer(DrillConfig config) throws DrillConfigurationException{
Optimizer o = config.getInstanceOf(OPTIMIZER_IMPL_KEY, Optimizer.class);
o.init(config);
return o;
}
-
+
public interface OptimizationContext{
public int getPriority();
+ public OptionManager getOptions();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PhysicalPlanReader.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PhysicalPlanReader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PhysicalPlanReader.java
index 167a992..8d77136 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PhysicalPlanReader.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PhysicalPlanReader.java
@@ -30,14 +30,14 @@ import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.physical.base.PhysicalOperatorUtil;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
import org.apache.drill.exec.record.MajorTypeSerDe;
+import org.apache.drill.exec.server.options.OptionList;
+import org.apache.drill.exec.store.StoragePluginRegistry;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.InjectableValues;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectReader;
import com.fasterxml.jackson.databind.module.SimpleModule;
-import org.apache.drill.exec.server.DrillbitContext;
-import org.apache.drill.exec.store.StoragePluginRegistry;
public class PhysicalPlanReader {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PhysicalPlanReader.class);
@@ -56,8 +56,8 @@ public class PhysicalPlanReader {
.addDeserializer(DrillbitEndpoint.class, new DrillbitEndpointSerDe.De()) //
.addSerializer(MajorType.class, new MajorTypeSerDe.Se())
.addDeserializer(MajorType.class, new MajorTypeSerDe.De());
-
-
+
+
mapper.registerModule(deserModule);
mapper.registerSubtypes(PhysicalOperatorUtil.getSubTypes(config));
InjectableValues injectables = new InjectableValues.Std() //
@@ -78,10 +78,14 @@ public class PhysicalPlanReader {
this(config, mapper, endpoint, null);
}
+ public String writeJson(OptionList list) throws JsonProcessingException{
+ return mapper.writeValueAsString(list);
+ }
+
public String writeJson(PhysicalOperator op) throws JsonProcessingException{
return mapper.writeValueAsString(op);
}
-
+
public PhysicalPlan readPhysicalPlan(String json) throws JsonProcessingException, IOException {
logger.debug("Reading physical plan {}", json);
return physicalPlanReader.readValue(json);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/planner/SimpleExecPlanner.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/SimpleExecPlanner.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/SimpleExecPlanner.java
index dff2dd3..4da6500 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/SimpleExecPlanner.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/SimpleExecPlanner.java
@@ -27,6 +27,7 @@ import org.apache.drill.exec.planner.fragment.MakeFragmentsVisitor;
import org.apache.drill.exec.planner.fragment.PlanningSet;
import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
import org.apache.drill.exec.planner.fragment.StatsCollector;
+import org.apache.drill.exec.server.options.OptionList;
import org.apache.drill.exec.work.QueryWorkUnit;
/**
@@ -34,25 +35,25 @@ import org.apache.drill.exec.work.QueryWorkUnit;
*/
public class SimpleExecPlanner implements ExecPlanner{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SimpleExecPlanner.class);
-
+
private MakeFragmentsVisitor fragmenter = new MakeFragmentsVisitor();
private SimpleParallelizer parallelizer = new SimpleParallelizer();
@Override
public QueryWorkUnit getWorkUnit(QueryContext context, PhysicalPlan plan, int maxWidth) throws ExecutionSetupException {
-
+
// get the root physical operator and split the plan into sub fragments.
PhysicalOperator root = plan.getSortedOperators(false).iterator().next();
Fragment fragmentRoot = root.accept(fragmenter, null);
-
+
// generate a planning set and collect stats.
PlanningSet planningSet = StatsCollector.collectStats(fragmentRoot);
int maxWidthPerEndpoint = context.getConfig().getInt(ExecConstants.MAX_WIDTH_PER_ENDPOINT);
-
- return parallelizer.getFragments(context.getCurrentEndpoint(), context.getQueryId(), context.getActiveEndpoints(),
+
+ return parallelizer.getFragments(new OptionList(), context.getCurrentEndpoint(), context.getQueryId(), context.getActiveEndpoints(),
context.getPlanReader(), fragmentRoot, planningSet, maxWidth, maxWidthPerEndpoint);
-
-
+
+
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizer.java
index 6e951df..c34869d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizer.java
@@ -23,6 +23,7 @@ import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.common.exceptions.PhysicalOperatorSetupException;
import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.expr.fn.impl.DateUtility;
import org.apache.drill.exec.physical.base.FragmentRoot;
import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.planner.PhysicalPlanReader;
@@ -31,6 +32,7 @@ import org.apache.drill.exec.proto.BitControl.PlanFragment;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.server.options.OptionList;
import org.apache.drill.exec.work.QueryWorkUnit;
import com.fasterxml.jackson.core.JsonProcessingException;
@@ -67,13 +69,13 @@ public class SimpleParallelizer {
* @return The list of generated PlanFragment protobuf objects to be assigned out to the individual nodes.
* @throws ExecutionSetupException
*/
- public QueryWorkUnit getFragments(DrillbitEndpoint foremanNode, QueryId queryId, Collection<DrillbitEndpoint> activeEndpoints, PhysicalPlanReader reader, Fragment rootNode, PlanningSet planningSet,
+ public QueryWorkUnit getFragments(OptionList options, DrillbitEndpoint foremanNode, QueryId queryId, Collection<DrillbitEndpoint> activeEndpoints, PhysicalPlanReader reader, Fragment rootNode, PlanningSet planningSet,
int globalMaxWidth, int maxWidthPerEndpoint) throws ExecutionSetupException {
assignEndpoints(activeEndpoints, planningSet, globalMaxWidth, maxWidthPerEndpoint);
- return generateWorkUnit(foremanNode, queryId, reader, rootNode, planningSet);
+ return generateWorkUnit(options, foremanNode, queryId, reader, rootNode, planningSet);
}
- private QueryWorkUnit generateWorkUnit(DrillbitEndpoint foremanNode, QueryId queryId, PhysicalPlanReader reader, Fragment rootNode,
+ private QueryWorkUnit generateWorkUnit(OptionList options, DrillbitEndpoint foremanNode, QueryId queryId, PhysicalPlanReader reader, Fragment rootNode,
PlanningSet planningSet) throws ExecutionSetupException {
List<PlanFragment> fragments = Lists.newArrayList();
@@ -109,8 +111,10 @@ public class SimpleParallelizer {
// get plan as JSON
String plan;
+ String optionsData;
try {
plan = reader.writeJson(root);
+ optionsData = reader.writeJson(options);
} catch (JsonProcessingException e) {
throw new FragmentSetupException("Failure while trying to convert fragment into json.", e);
}
@@ -135,6 +139,7 @@ public class SimpleParallelizer {
.setTimeZone(timeZone)//
.setMemInitial(wrapper.getInitialAllocation())//
.setMemMax(wrapper.getMaxAllocation())
+ .setOptionsJson(optionsData)
.build();
if (isRootNode) {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/StoragePlugins.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/StoragePlugins.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/StoragePlugins.java
index 939b77c..d296eb1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/StoragePlugins.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/StoragePlugins.java
@@ -26,21 +26,22 @@ import java.util.Map.Entry;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.logical.StoragePluginConfig;
+import org.apache.drill.exec.cache.JacksonSerializable;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Charsets;
import com.google.common.io.Resources;
-public class StoragePlugins implements Iterable<Map.Entry<String, StoragePluginConfig>>{
-
+public class StoragePlugins extends JacksonSerializable implements Iterable<Map.Entry<String, StoragePluginConfig>>{
+
private Map<String, StoragePluginConfig> storage;
-
+
@JsonCreator
public StoragePlugins(@JsonProperty("storage") Map<String, StoragePluginConfig> storage){
this.storage = storage;
}
-
+
public static void main(String[] args) throws Exception{
DrillConfig config = DrillConfig.create();
String data = Resources.toString(Resources.getResource("storage-engines.json"), Charsets.UTF_8);
@@ -88,6 +89,6 @@ public class StoragePlugins implements Iterable<Map.Entry<String, StoragePluginC
}
return storage.equals(((StoragePlugins) obj).getStorage());
}
-
-
+
+
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java
index e6e99c0..5e49451 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java
@@ -19,17 +19,24 @@ package org.apache.drill.exec.planner.physical;
import net.hydromatic.optiq.tools.FrameworkContext;
+import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.exec.server.options.OptionValidator;
+import org.apache.drill.exec.server.options.TypeValidators.BooleanValidator;
+
public class PlannerSettings implements FrameworkContext{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PlannerSettings.class);
- private boolean singleMode;
- public boolean isSingleMode() {
- return singleMode;
+ public static final OptionValidator EXCHANGE = new BooleanValidator("planner.disable_exchanges", false);
+
+ public OptionManager options;
+
+ public PlannerSettings(OptionManager options){
+ this.options = options;
}
- public void setSingleMode(boolean singleMode) {
- this.singleMode = singleMode;
+ public boolean isSingleMode() {
+ return options.getOption(EXCHANGE.getOptionName()).bool_val;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java
index 0e9f798..116017c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java
@@ -25,6 +25,7 @@ import net.hydromatic.optiq.tools.Planner;
import net.hydromatic.optiq.tools.RelConversionException;
import net.hydromatic.optiq.tools.ValidationException;
+import org.apache.drill.common.JSONOptions;
import org.apache.drill.common.logical.PlanProperties;
import org.apache.drill.common.logical.PlanProperties.Generator.ResultMode;
import org.apache.drill.common.logical.PlanProperties.PlanPropertiesBuilder;
@@ -34,12 +35,10 @@ import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.AbstractPhysicalVisitor;
import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.planner.logical.DrillRel;
-import org.apache.drill.exec.planner.logical.DrillRuleSets;
import org.apache.drill.exec.planner.logical.DrillScreenRel;
import org.apache.drill.exec.planner.logical.DrillStoreRel;
import org.apache.drill.exec.planner.physical.DrillDistributionTrait;
import org.apache.drill.exec.planner.physical.PhysicalPlanCreator;
-import org.apache.drill.exec.planner.physical.PlannerSettings;
import org.apache.drill.exec.planner.physical.Prel;
import org.apache.drill.exec.planner.sql.DrillSqlWorker;
import org.eigenbase.rel.RelNode;
@@ -132,6 +131,7 @@ public class DefaultSqlHandler implements SqlHandler{
PlanPropertiesBuilder propsBuilder = PlanProperties.builder();
propsBuilder.type(PlanType.APACHE_DRILL_PHYSICAL);
propsBuilder.version(1);
+ propsBuilder.options(new JSONOptions(context.getOptions().getSessionOptionList()));
propsBuilder.resultMode(ResultMode.EXEC);
propsBuilder.generator(this.getClass().getSimpleName(), "");
return new PhysicalPlan(propsBuilder.build(), getPops(op));
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ExplainHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ExplainHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ExplainHandler.java
index dec9222..86ce6c5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ExplainHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ExplainHandler.java
@@ -59,7 +59,7 @@ public class ExplainHandler extends DefaultSqlHandler{
log("Drill Logical", drel);
if(mode == ResultMode.LOGICAL){
- LogicalExplain logicalResult = new LogicalExplain(drel);
+ LogicalExplain logicalResult = new LogicalExplain(drel, level, context);
return DirectPlan.createDirectPlan(context, logicalResult);
}
@@ -68,7 +68,7 @@ public class ExplainHandler extends DefaultSqlHandler{
PhysicalOperator pop = convertToPop(prel);
PhysicalPlan plan = convertToPlan(pop);
log("Drill Plan", plan);
- PhysicalExplain physicalResult = new PhysicalExplain(prel, plan);
+ PhysicalExplain physicalResult = new PhysicalExplain(prel, plan, level, context);
return DirectPlan.createDirectPlan(context, physicalResult);
}
@@ -93,11 +93,11 @@ public class ExplainHandler extends DefaultSqlHandler{
}
- public class LogicalExplain{
+ public static class LogicalExplain{
public String text;
public String json;
- public LogicalExplain(RelNode node){
+ public LogicalExplain(RelNode node, SqlExplainLevel level, QueryContext context){
this.text = RelOptUtil.toString(node, level);
DrillImplementor implementor = new DrillImplementor(new DrillParseContext(), ResultMode.LOGICAL);
implementor.go( (DrillRel) node);
@@ -106,11 +106,11 @@ public class ExplainHandler extends DefaultSqlHandler{
}
}
- public class PhysicalExplain{
+ public static class PhysicalExplain{
public String text;
public String json;
- public PhysicalExplain(RelNode node, PhysicalPlan plan){
+ public PhysicalExplain(RelNode node, PhysicalPlan plan, SqlExplainLevel level, QueryContext context){
this.text = RelOptUtil.toString(node, level);
this.json = plan.unparse(context.getConfig().getMapper().writer());
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SetOptionHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SetOptionHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SetOptionHandler.java
index 64be891..ae3960a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SetOptionHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SetOptionHandler.java
@@ -25,6 +25,7 @@ import net.hydromatic.optiq.tools.ValidationException;
import org.apache.drill.exec.ops.QueryContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.planner.sql.DirectPlan;
+import org.eigenbase.sql.SqlLiteral;
import org.eigenbase.sql.SqlNode;
import org.eigenbase.sql.SqlSetOption;
@@ -46,10 +47,23 @@ public class SetOptionHandler implements SqlHandler{
String scope = option.getScope();
String name = option.getName();
SqlNode value = option.getValue();
- if(name.equals("NO_EXCHANGES")){
- context.getPlannerSettings().setSingleMode(true);
+ if(value instanceof SqlLiteral){
+ switch(scope.toLowerCase()){
+ case "session":
+ context.getOptions().setOption(option.getName(), (SqlLiteral) value);
+ break;
+ case "system":
+ context.getOptions().getSystemManager().setOption(name, (SqlLiteral) value);
+ break;
+ default:
+ throw new ValidationException("Invalid OPTION scope. Scope must be SESSION or SYSTEM.");
+ }
+
+ }else{
+ throw new ValidationException("Sql options can only be literals.");
}
- return DirectPlan.createDirectPlan(context, true, "disabled exchanges.");
+
+ return DirectPlan.createDirectPlan(context, true, String.format("%s updated.", name));
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SimpleCommandResult.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SimpleCommandResult.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SimpleCommandResult.java
index 6dd7e7b..07d26b4 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SimpleCommandResult.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/SimpleCommandResult.java
@@ -18,7 +18,6 @@
package org.apache.drill.exec.planner.sql.handlers;
public class SimpleCommandResult {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SimpleCommandResult.class);
public boolean ok;
public String summary;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
index acd8412..3ee25d9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
@@ -17,13 +17,13 @@
*/
package org.apache.drill.exec.rpc.user;
-import java.io.IOException;
-
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
+import java.io.IOException;
+
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.physical.impl.materialize.QueryWritableBatch;
import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
@@ -100,12 +100,13 @@ public class UserServer extends BasicServer<RpcType, UserServer.UserClientConnec
public class UserClientConnection extends RemoteConnection {
private UserSession session;
+
public UserClientConnection(Channel channel) {
super(channel);
}
void setUser(UserCredentials credentials, UserProperties props) throws IOException{
- session = new UserSession(credentials, props);
+ session = new UserSession(worker.getSystemOptions(), credentials, props);
}
public UserSession getSession(){
@@ -114,7 +115,6 @@ public class UserServer extends BasicServer<RpcType, UserServer.UserClientConnec
public void sendResult(RpcOutcomeListener<Ack> listener, QueryWritableBatch result){
// logger.debug("Sending result to client with {}", result);
-
send(listener, this, RpcType.QUERY_RESULT, result.getHeader(), Ack.class, result.getBuffers());
}
@@ -122,6 +122,8 @@ public class UserServer extends BasicServer<RpcType, UserServer.UserClientConnec
public BufferAllocator getAllocator() {
return alloc;
}
+
+
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java
index f27317c..86c4bad 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java
@@ -18,14 +18,18 @@
package org.apache.drill.exec.rpc.user;
import java.io.IOException;
+import java.util.Iterator;
import java.util.Map;
-import com.google.common.collect.Maps;
import net.hydromatic.optiq.SchemaPlus;
import org.apache.drill.exec.proto.UserBitShared.UserCredentials;
import org.apache.drill.exec.proto.UserProtos.Property;
import org.apache.drill.exec.proto.UserProtos.UserProperties;
+import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.exec.server.options.SessionOptionManager;
+
+import com.google.common.collect.Maps;
public class UserSession {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UserSession.class);
@@ -36,10 +40,11 @@ public class UserSession {
private boolean enableExchanges = true;
private UserCredentials credentials;
private Map<String, String> properties;
+ private OptionManager options;
- public UserSession(UserCredentials credentials, UserProperties properties) throws IOException{
+ public UserSession(OptionManager systemOptions, UserCredentials credentials, UserProperties properties) throws IOException{
this.credentials = credentials;
-
+ this.options = new SessionOptionManager(systemOptions);
this.properties = Maps.newHashMap();
if (properties == null) return;
for (int i=0; i<properties.getPropertiesCount(); i++) {
@@ -48,6 +53,10 @@ public class UserSession {
}
}
+ public OptionManager getOptions(){
+ return options;
+ }
+
public DrillUser getUser(){
return user;
}
@@ -95,4 +104,5 @@ public class UserSession {
}
return schema;
}
+
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/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 411a76c..7f607a3 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
@@ -22,6 +22,7 @@ import java.io.Closeable;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.cache.DistributedCache;
+import org.apache.drill.exec.cache.DistributedMultiMap;
import org.apache.drill.exec.cache.HazelCache;
import org.apache.drill.exec.coord.ClusterCoordinator;
import org.apache.drill.exec.coord.ClusterCoordinator.RegistrationHandle;
@@ -96,6 +97,7 @@ public class Drillbit implements Closeable{
manager.start(md, cache, engine.getController(), engine.getDataConnectionCreator(), coord);
cache.run();
manager.getContext().getStorage().init();
+ manager.getContext().getOptionManager().init();
handle = coord.register(md);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java.orig
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java.orig b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java.orig
new file mode 100644
index 0000000..5e72717
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java.orig
@@ -0,0 +1,141 @@
+/**
+ * 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.server;
+
+import java.io.Closeable;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.cache.DistributedCache;
+import org.apache.drill.exec.cache.DistributedMultiMap;
+import org.apache.drill.exec.cache.HazelCache;
+import org.apache.drill.exec.coord.ClusterCoordinator;
+import org.apache.drill.exec.coord.ClusterCoordinator.RegistrationHandle;
+import org.apache.drill.exec.coord.ZKClusterCoordinator;
+import org.apache.drill.exec.exception.DrillbitStartupException;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.service.ServiceEngine;
+import org.apache.drill.exec.work.WorkManager;
+
+import com.google.common.io.Closeables;
+
+/**
+ * Starts, tracks and stops all the required services for a Drillbit daemon to work.
+ */
+public class Drillbit implements Closeable{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(Drillbit.class);
+
+ public static Drillbit start(StartupOptions options) throws DrillbitStartupException {
+ return start(DrillConfig.create(options.getConfigLocation()));
+ }
+
+ public static Drillbit start(DrillConfig config) throws DrillbitStartupException {
+ Drillbit bit;
+ try {
+ logger.debug("Setting up Drillbit.");
+ bit = new Drillbit(config, null);
+ } catch (Exception ex) {
+ throw new DrillbitStartupException("Failure while initializing values in Drillbit.", ex);
+ }
+ try {
+ logger.debug("Starting Drillbit.");
+ bit.run();
+ } catch (Exception e) {
+ throw new DrillbitStartupException("Failure during initial startup of Drillbit.", e);
+ }
+ return bit;
+ }
+
+ public static void main(String[] cli) throws DrillbitStartupException {
+ StartupOptions options = StartupOptions.parse(cli);
+ start(options);
+ }
+
+ final ClusterCoordinator coord;
+ final ServiceEngine engine;
+ final DistributedCache cache;
+ final WorkManager manager;
+ final BootStrapContext context;
+
+ private volatile RegistrationHandle handle;
+
+ public Drillbit(DrillConfig config, RemoteServiceSet serviceSet) throws Exception {
+ if(serviceSet != null){
+ this.context = new BootStrapContext(config);
+ this.manager = new WorkManager(context);
+ this.coord = serviceSet.getCoordinator();
+ this.engine = new ServiceEngine(manager.getControlMessageHandler(), manager.getUserWorker(), context, manager.getWorkBus(), manager.getDataHandler());
+ this.cache = serviceSet.getCache();
+ }else{
+ Runtime.getRuntime().addShutdownHook(new ShutdownThread(config));
+ this.context = new BootStrapContext(config);
+ this.manager = new WorkManager(context);
+ this.coord = new ZKClusterCoordinator(config);
+ this.engine = new ServiceEngine(manager.getControlMessageHandler(), manager.getUserWorker(), context, manager.getWorkBus(), manager.getDataHandler());
+ this.cache = new HazelCache(config, context.getAllocator());
+ }
+ }
+
+ public void run() throws Exception {
+ coord.start(10000);
+ DrillbitEndpoint md = engine.start();
+ cache.run();
+<<<<<<< HEAD
+ manager.getContext().getStorage().init();
+=======
+ manager.start(md, cache, engine.getController(), engine.getDataConnectionCreator(), coord);
+>>>>>>> Drill 381 - implementation of session and global options.
+ handle = coord.register(md);
+ }
+
+ public void close() {
+ if (coord != null) coord.unregister(handle);
+
+ try {
+ Thread.sleep(context.getConfig().getInt(ExecConstants.ZK_REFRESH) * 2);
+ } catch (InterruptedException e) {
+ logger.warn("Interrupted while sleeping during coordination deregistration.");
+ }
+
+ Closeables.closeQuietly(engine);
+ Closeables.closeQuietly(coord);
+ Closeables.closeQuietly(manager);
+ Closeables.closeQuietly(context);
+ logger.info("Shutdown completed.");
+ }
+
+ private class ShutdownThread extends Thread {
+ ShutdownThread(DrillConfig config) {
+ this.setName("ShutdownHook");
+ }
+
+ @Override
+ public void run() {
+ logger.info("Received shutdown request.");
+ close();
+ }
+
+ }
+ public ClusterCoordinator getCoordinator(){
+ return coord;
+ }
+
+ public DrillbitContext getContext(){
+ return this.manager.getContext();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/08923cb8/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 1fd30e3..f0a47f8 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
@@ -22,7 +22,6 @@ import io.netty.channel.nio.NioEventLoopGroup;
import java.util.Collection;
import org.apache.drill.common.config.DrillConfig;
-import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.exec.cache.DistributedCache;
import org.apache.drill.exec.coord.ClusterCoordinator;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
@@ -33,9 +32,10 @@ import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
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.OptionManager;
+import org.apache.drill.exec.server.options.SystemOptionManager;
import org.apache.drill.exec.store.StoragePluginRegistry;
import org.apache.drill.exec.store.StoragePluginRegistry.DrillSchemaFactory;
-import org.apache.drill.exec.store.StoragePlugin;
import com.codahale.metrics.MetricRegistry;
import com.google.common.base.Preconditions;
@@ -55,6 +55,7 @@ public class DrillbitContext {
private final Controller controller;
private final WorkEventBus workBus;
private final FunctionImplementationRegistry functionRegistry;
+ private final SystemOptionManager systemOptions;
public DrillbitContext(DrillbitEndpoint endpoint, BootStrapContext context, ClusterCoordinator coord, Controller controller, DataConnectionCreator connectionsPool, DistributedCache cache, WorkEventBus workBus) {
super();
@@ -73,6 +74,9 @@ public class DrillbitContext {
this.reader = new PhysicalPlanReader(context.getConfig(), context.getConfig().getMapper(), endpoint, storagePlugins);
this.operatorCreatorRegistry = new OperatorCreatorRegistry(context.getConfig());
this.functionRegistry = new FunctionImplementationRegistry(context.getConfig());
+ this.systemOptions = new SystemOptionManager(cache);
+
+// this.globalDrillOptions = new DistributedGlobalOptions(this.cache);
}
public FunctionImplementationRegistry getFunctionImplementationRegistry() {
@@ -83,6 +87,10 @@ public class DrillbitContext {
return workBus;
}
+ public SystemOptionManager getOptionManager() {
+ return systemOptions;
+ }
+
public DrillbitEndpoint getEndpoint(){
return endpoint;
}