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 2014/06/11 05:51:58 UTC
[13/61] [abbrv] Enable View persistence,
Storage Plugin and System option persistence.
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java
index 2f41c26..e61cd76 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java
@@ -21,7 +21,9 @@ import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
-import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
@@ -38,7 +40,6 @@ import org.apache.drill.exec.cache.DistributedCache;
import org.apache.drill.exec.cache.DistributedMap;
import org.apache.drill.exec.cache.DistributedMultiMap;
import org.apache.drill.exec.cache.DrillSerializable;
-import org.apache.drill.exec.cache.JacksonSerializable;
import org.apache.drill.exec.exception.DrillbitStartupException;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.memory.TopLevelAllocator;
@@ -46,20 +47,22 @@ 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.base.Charsets;
import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.ListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
-import com.google.common.io.ByteArrayDataInput;
+import com.google.common.collect.Multimaps;
import com.google.common.io.ByteArrayDataOutput;
import com.google.common.io.ByteStreams;
+import com.google.protobuf.Message;
+import com.google.protobuf.Parser;
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<CacheConfig<?, ?>, DistributedMap<?, ?>> maps;
+ private volatile ConcurrentMap<CacheConfig<?, ?>, DistributedMultiMap<?, ?>> multiMaps;
private volatile ConcurrentMap<String, Counter> counters;
private static final BufferAllocator allocator = new TopLevelAllocator(DrillConfig.create());
@@ -67,59 +70,33 @@ public class LocalCache implements DistributedCache {
@Override
public void close() throws IOException {
- handles = null;
+
}
@Override
public void run() throws DrillbitStartupException {
- handles = Maps.newConcurrentMap();
maps = Maps.newConcurrentMap();
multiMaps = Maps.newConcurrentMap();
counters = Maps.newConcurrentMap();
- namedMaps = Maps.newConcurrentMap();
- }
-
- @Override
- public PlanFragment getFragment(FragmentHandle handle) {
-// logger.debug("looking for fragment with handle: {}", handle);
- return handles.get(handle);
}
@Override
- public void storeFragment(PlanFragment fragment) {
-// 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);
+ public <K, V> DistributedMultiMap<K, V> getMultiMap(CacheConfig<K, V> config) {
+ DistributedMultiMap<K, V> mmap = (DistributedMultiMap<K, V>) multiMaps.get(config);
if (mmap == null) {
- multiMaps.putIfAbsent(clazz, new LocalDistributedMultiMapImpl<V>(clazz));
- return (DistributedMultiMap<V>) multiMaps.get(clazz);
+ multiMaps.putIfAbsent(config, new LocalDistributedMultiMapImpl<K, V>(config));
+ return (DistributedMultiMap<K, V>) multiMaps.get(config);
} else {
return mmap;
}
}
@Override
- public <V extends DrillSerializable> DistributedMap<V> getMap(Class<V> clazz) {
- DistributedMap m = maps.get(clazz);
- if (m == null) {
- maps.putIfAbsent(clazz, new LocalDistributedMapImpl<V>(clazz));
- return (DistributedMap<V>) maps.get(clazz);
- } else {
- return m;
- }
- }
-
-
- @Override
- public <V extends DrillSerializable> DistributedMap<V> getNamedMap(String name, Class<V> clazz) {
- DistributedMap m = namedMaps.get(clazz);
+ public <K, V> DistributedMap<K, V> getMap(CacheConfig<K, V> config) {
+ DistributedMap<K, V> m = (DistributedMap<K, V>) maps.get(config);
if (m == null) {
- namedMaps.putIfAbsent(name, new LocalDistributedMapImpl<V>(clazz));
- return (DistributedMap<V>) namedMaps.get(name);
+ maps.putIfAbsent(config, new LocalDistributedMapImpl<K, V>(config));
+ return (DistributedMap<K, V>) maps.get(config);
} else {
return m;
}
@@ -136,112 +113,186 @@ public class LocalCache implements DistributedCache {
}
}
- public static ByteArrayDataOutput serialize(DrillSerializable obj) {
- if(obj instanceof JacksonSerializable){
+ private static BytesHolder serialize(Object obj, SerializationMode mode) {
+ if(obj instanceof String){
+ return new BytesHolder( ((String)obj).getBytes(Charsets.UTF_8));
+ }
try{
+ switch(mode){
+
+ case DRILL_SERIALIZIABLE: {
+ ByteArrayDataOutput out = ByteStreams.newDataOutput();
+ OutputStream outputStream = DataOutputOutputStream.constructOutputStream(out);
+ ((DrillSerializable)obj).writeToStream(outputStream);
+ outputStream.flush();
+ return new BytesHolder(out.toByteArray());
+ }
+
+ case JACKSON: {
ByteArrayDataOutput out = ByteStreams.newDataOutput();
out.write(mapper.writeValueAsBytes(obj));
- return out;
- }catch(Exception e){
- throw new RuntimeException(e);
+ return new BytesHolder(out.toByteArray());
}
- }
- ByteArrayDataOutput out = ByteStreams.newDataOutput();
- OutputStream outputStream = DataOutputOutputStream.constructOutputStream(out);
- try {
- obj.writeToStream(outputStream);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- try {
- outputStream.flush();
- } catch (IOException e) {
+ case PROTOBUF:
+ return new BytesHolder(( (Message) obj).toByteArray());
+
+ }
+ }catch(Exception e){
throw new RuntimeException(e);
}
- return out;
+
+ throw new UnsupportedOperationException();
}
- public static <V extends DrillSerializable> V deserialize(byte[] bytes, Class<V> clazz) {
- if(JacksonSerializable.class.isAssignableFrom(clazz)){
- try{
+ private static <V> V deserialize(BytesHolder b, SerializationMode mode, Class<V> clazz) {
+ byte[] bytes = b.bytes;
+ try{
+
+ if (clazz == String.class) {
+ return (V) new String(bytes, Charsets.UTF_8);
+ }
+
+ switch (mode) {
+
+ case DRILL_SERIALIZIABLE: {
+ InputStream inputStream = new ByteArrayInputStream(bytes);
+ V obj = clazz.getConstructor(BufferAllocator.class).newInstance(allocator);
+ ((DrillSerializable) obj).readFromStream(inputStream);
+ return obj;
+ }
+
+ case JACKSON: {
return (V) mapper.readValue(bytes, clazz);
- }catch(Exception e){
- throw new RuntimeException(e);
}
- }
- InputStream inputStream = new ByteArrayInputStream(bytes);
- try {
- V obj = clazz.getConstructor(BufferAllocator.class).newInstance(allocator);
- obj.readFromStream(inputStream);
- return obj;
- } catch (InstantiationException | IllegalAccessException | IOException | NoSuchMethodException | InvocationTargetException e) {
+ case PROTOBUF: {
+ Parser<V> parser = null;
+ for (Field f : clazz.getFields()) {
+ if (f.getName().equals("PARSER") && Modifier.isStatic(f.getModifiers())) {
+ parser = (Parser<V>) f.get(null);
+ }
+ }
+ if (parser == null) throw new UnsupportedOperationException(String.format("Unable to find parser for class %s.", clazz.getName()));
+ InputStream inputStream = new ByteArrayInputStream(bytes);
+ return parser.parseFrom(inputStream);
+ }
+
+ }
+ }catch(Exception e){
throw new RuntimeException(e);
}
+
+ throw new UnsupportedOperationException();
+ }
+
+ private static class BytesHolder {
+ final byte[] bytes;
+ public BytesHolder(byte[] bytes){
+ this.bytes = bytes;
+ }
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + Arrays.hashCode(bytes);
+ return result;
+ }
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ BytesHolder other = (BytesHolder) obj;
+ if (!Arrays.equals(bytes, other.bytes))
+ return false;
+ return true;
+ }
+
+
}
- public static class LocalDistributedMultiMapImpl<V extends DrillSerializable> implements DistributedMultiMap<V> {
- private ArrayListMultimap<String, ByteArrayDataOutput> mmap;
- private Class<V> clazz;
+ static class LocalDistributedMultiMapImpl<K, V> implements DistributedMultiMap<K, V> {
+ private ListMultimap<BytesHolder, BytesHolder> mmap;
+ private CacheConfig<K, V> config;
- public LocalDistributedMultiMapImpl(Class<V> clazz) {
- mmap = ArrayListMultimap.create();
- this.clazz = clazz;
+ public LocalDistributedMultiMapImpl(CacheConfig<K, V> config) {
+ ListMultimap<BytesHolder, BytesHolder> innerMap = ArrayListMultimap.create();
+ mmap = Multimaps.synchronizedListMultimap(innerMap);
+ this.config = config;
}
@Override
- public Collection<V> get(String key) {
+ public Collection<V> get(K key) {
List<V> list = Lists.newArrayList();
- for (ByteArrayDataOutput o : mmap.get(key)) {
- list.add(deserialize(o.toByteArray(), this.clazz));
+ for (BytesHolder o : mmap.get(serialize(key, config.getMode()))) {
+ list.add(deserialize(o, config.getMode(), config.getValueClass()));
}
return list;
}
@Override
- public void put(String key, DrillSerializable value) {
- mmap.put(key, serialize(value));
+ public void put(K key, V value) {
+ mmap.put(serialize(key, config.getMode()), serialize(value, config.getMode()));
}
}
- public static class LocalDistributedMapImpl<V extends DrillSerializable> implements DistributedMap<V> {
- protected ConcurrentMap<String, ByteArrayDataOutput> m;
- protected Class<V> clazz;
+ public static class LocalDistributedMapImpl<K, V> implements DistributedMap<K, V> {
+ protected ConcurrentMap<BytesHolder, BytesHolder> m;
+ protected CacheConfig<K, V> config;
- public LocalDistributedMapImpl(Class<V> clazz) {
+ public LocalDistributedMapImpl(CacheConfig<K, V> config) {
m = Maps.newConcurrentMap();
- this.clazz = clazz;
+ this.config = config;
+ }
+
+ @Override
+ public V get(K key) {
+ BytesHolder b = m.get(serialize(key, config.getMode()));
+ if(b == null) return null;
+ return (V) deserialize(b, config.getMode(), config.getValueClass());
}
@Override
- public V get(String key) {
- if (m.get(key) == null) return null;
- ByteArrayDataOutput b = m.get(key);
- byte[] bytes = b.toByteArray();
- return (V) deserialize(bytes, this.clazz);
+ public Iterable<Entry<K, V>> getLocalEntries() {
+ return new Iterable<Entry<K, V>>(){
+ @Override
+ public Iterator<Entry<K, V>> iterator() {
+ return new DeserializingTransformer(m.entrySet().iterator());
+ }
+ };
+
}
@Override
- public void put(String key, V value) {
- m.put(key, serialize(value));
+ public void put(K key, V value) {
+ m.put(serialize(key, config.getMode()), serialize(value, config.getMode()));
}
+
@Override
- public void putIfAbsent(String key, V value) {
- m.putIfAbsent(key, serialize(value));
+ public void putIfAbsent(K key, V value) {
+ m.putIfAbsent(serialize(key, config.getMode()), serialize(value, config.getMode()));
}
@Override
- public void putIfAbsent(String key, V value, long ttl, TimeUnit timeUnit) {
- m.putIfAbsent(key, serialize(value));
+ public void delete(K key) {
+ m.remove(serialize(key, config.getMode()));
+ }
+
+ @Override
+ public void putIfAbsent(K key, V value, long ttl, TimeUnit timeUnit) {
+ m.putIfAbsent(serialize(key, config.getMode()), serialize(value, config.getMode()));
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;
+ private class DeserializingTransformer implements Iterator<Map.Entry<K, V>> {
+ private Iterator<Map.Entry<BytesHolder, BytesHolder>> inner;
- public DeserializingTransformer(Iterator<Entry<String, ByteArrayDataOutput>> inner) {
+ public DeserializingTransformer(Iterator<Entry<BytesHolder, BytesHolder>> inner) {
super();
this.inner = inner;
}
@@ -252,7 +303,7 @@ public class LocalCache implements DistributedCache {
}
@Override
- public Entry<String, V> next() {
+ public Entry<K, V> next() {
return newEntry(inner.next());
}
@@ -261,17 +312,17 @@ public class LocalCache implements DistributedCache {
throw new UnsupportedOperationException();
}
- public Entry<String, V> newEntry(final Entry<String, ByteArrayDataOutput> input) {
- return new Map.Entry<String, V>(){
+ public Entry<K, V> newEntry(final Entry<BytesHolder, BytesHolder> input) {
+ return new Map.Entry<K, V>() {
@Override
- public String getKey() {
- return input.getKey();
+ public K getKey() {
+ return deserialize(input.getKey(), config.getMode(), config.getKeyClass());
}
@Override
public V getValue() {
- return deserialize(input.getValue().toByteArray(), clazz);
+ return deserialize(input.getValue(), config.getMode(), config.getValueClass());
}
@Override
@@ -283,10 +334,8 @@ public class LocalCache implements DistributedCache {
}
}
- @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/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
index 9cd2cdd..7dc7702 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
@@ -33,7 +33,7 @@ import java.util.Vector;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.coord.ClusterCoordinator;
-import org.apache.drill.exec.coord.ZKClusterCoordinator;
+import org.apache.drill.exec.coord.zk.ZKClusterCoordinator;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.memory.TopLevelAllocator;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java
index 4153a24..2f5c027 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java
@@ -23,7 +23,7 @@ import java.nio.file.Paths;
import java.util.concurrent.TimeUnit;
import org.apache.drill.common.config.DrillConfig;
-import org.apache.drill.exec.coord.ZKClusterCoordinator;
+import org.apache.drill.exec.coord.zk.ZKClusterCoordinator;
import org.apache.drill.exec.proto.UserBitShared.QueryType;
import org.apache.drill.exec.server.Drillbit;
import org.apache.drill.exec.server.RemoteServiceSet;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/coord/ClusterCoordinator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/ClusterCoordinator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/ClusterCoordinator.java
index b832e31..f855107 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/ClusterCoordinator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/ClusterCoordinator.java
@@ -20,6 +20,7 @@ package org.apache.drill.exec.coord;
import java.io.Closeable;
import java.util.Collection;
+import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
/**
@@ -30,7 +31,7 @@ public abstract class ClusterCoordinator implements Closeable {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ClusterCoordinator.class);
/**
- * Start the cluster coordinator. Millis to wait is
+ * Start the cluster coordinator. Millis to wait is
* @param millisToWait The maximum time to wait before throwing an exception if the cluster coordination service has not successfully started. Use 0 to wait indefinitely.
* @throws Exception
*/
@@ -51,4 +52,7 @@ public abstract class ClusterCoordinator implements Closeable {
public interface RegistrationHandle {
}
+ public abstract DistributedSemaphore getSemaphore(String name, int maximumLeases);
+
+
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/coord/DistributedSemaphore.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/DistributedSemaphore.java b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/DistributedSemaphore.java
new file mode 100644
index 0000000..d9d8764
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/DistributedSemaphore.java
@@ -0,0 +1,26 @@
+/**
+ * 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.coord;
+
+import java.util.concurrent.TimeUnit;
+
+public interface DistributedSemaphore {
+ public DistributedLease acquire(long time, TimeUnit unit) throws Exception;
+
+ public interface DistributedLease extends AutoCloseable{}
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/coord/DrillServiceInstanceHelper.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/DrillServiceInstanceHelper.java b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/DrillServiceInstanceHelper.java
index 9573741..94b3a11 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/DrillServiceInstanceHelper.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/DrillServiceInstanceHelper.java
@@ -17,20 +17,19 @@
*/
package org.apache.drill.exec.coord;
+import org.apache.curator.x.discovery.ServiceInstance;
+import org.apache.curator.x.discovery.ServiceInstanceBuilder;
+import org.apache.curator.x.discovery.details.InstanceSerializer;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.proto.CoordinationProtos.DrillServiceInstance;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
-import com.netflix.curator.x.discovery.ServiceInstance;
-import com.netflix.curator.x.discovery.ServiceInstanceBuilder;
-import com.netflix.curator.x.discovery.details.InstanceSerializer;
-
public class DrillServiceInstanceHelper {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillServiceInstanceHelper.class);
-
+
public static final InstanceSerializer<DrillbitEndpoint> SERIALIZER = new DrillServiceInstanceSerializer();
-
+
private static class DrillServiceInstanceSerializer implements InstanceSerializer<DrillbitEndpoint>{
@Override
@@ -52,6 +51,6 @@ public class DrillServiceInstanceHelper {
b.payload(i.getEndpoint());
return b.build();
}
-
+
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/coord/LocalClusterCoordinator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/LocalClusterCoordinator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/LocalClusterCoordinator.java
deleted file mode 100644
index b9e54ca..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/LocalClusterCoordinator.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/**
- * 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.coord;
-
-import java.io.IOException;
-import java.util.Collection;
-import java.util.Map;
-import java.util.UUID;
-
-import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
-
-import com.google.common.collect.Maps;
-
-public class LocalClusterCoordinator extends ClusterCoordinator{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(LocalClusterCoordinator.class);
-
- private volatile Map<RegistrationHandle, DrillbitEndpoint> endpoints = Maps.newConcurrentMap();
-
- @Override
- public void close() throws IOException {
- endpoints.clear();
- }
-
- @Override
- public void start(long millis) throws Exception {
- logger.debug("Local Cluster Coordinator started.");
- }
-
- @Override
- public RegistrationHandle register(DrillbitEndpoint data) {
- logger.debug("Endpoint registered {}.", data);
- Handle h = new Handle();
- endpoints.put(h, data);
- return h;
- }
-
- @Override
- public void unregister(RegistrationHandle handle) {
- if(handle == null) return;
-
- endpoints.remove(handle);
- }
-
- @Override
- public Collection<DrillbitEndpoint> getAvailableEndpoints() {
- return endpoints.values();
- }
-
-
- private class Handle implements RegistrationHandle{
- UUID id = UUID.randomUUID();
-
- @Override
- public int hashCode() {
- final int prime = 31;
- int result = 1;
- result = prime * result + getOuterType().hashCode();
- result = prime * result + ((id == null) ? 0 : id.hashCode());
- return result;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) return true;
- if (obj == null) return false;
- if (getClass() != obj.getClass()) return false;
- Handle other = (Handle) obj;
- if (!getOuterType().equals(other.getOuterType())) return false;
- if (id == null) {
- if (other.id != null) return false;
- } else if (!id.equals(other.id)) return false;
- return true;
- }
-
- private LocalClusterCoordinator getOuterType() {
- return LocalClusterCoordinator.this;
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/coord/ZKClusterCoordinator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/ZKClusterCoordinator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/ZKClusterCoordinator.java
deleted file mode 100644
index de3ed76..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/ZKClusterCoordinator.java
+++ /dev/null
@@ -1,198 +0,0 @@
-/**
- * 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.coord;
-
-import static com.google.common.base.Throwables.propagate;
-import static com.google.common.collect.Collections2.transform;
-
-import java.io.IOException;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.drill.common.config.DrillConfig;
-import org.apache.drill.exec.ExecConstants;
-import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
-
-import com.google.common.base.Function;
-import com.netflix.curator.RetryPolicy;
-import com.netflix.curator.framework.CuratorFramework;
-import com.netflix.curator.framework.CuratorFrameworkFactory;
-import com.netflix.curator.framework.state.ConnectionState;
-import com.netflix.curator.framework.state.ConnectionStateListener;
-import com.netflix.curator.retry.RetryNTimes;
-import com.netflix.curator.x.discovery.ServiceDiscovery;
-import com.netflix.curator.x.discovery.ServiceDiscoveryBuilder;
-import com.netflix.curator.x.discovery.ServiceInstance;
-import com.netflix.curator.x.discovery.details.ServiceCache;
-import com.netflix.curator.x.discovery.details.ServiceCacheListener;
-
-/**
- * Manages cluster coordination utilizing zookeeper. *
- */
-public class ZKClusterCoordinator extends ClusterCoordinator {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ZKClusterCoordinator.class);
-
- private String basePath;
- private CuratorFramework curator;
- private ServiceDiscovery<DrillbitEndpoint> discovery;
- private ServiceCache<DrillbitEndpoint> serviceCache;
- private volatile Collection<DrillbitEndpoint> endpoints = Collections.emptyList();
- private final String serviceName;
- private final CountDownLatch initialConnection = new CountDownLatch(1);
-
-
- public ZKClusterCoordinator(DrillConfig config) throws IOException{
- this(config, null);
- }
-
- public ZKClusterCoordinator(DrillConfig config, String connect) throws IOException {
- connect = connect == null || connect.isEmpty() ? config.getString(ExecConstants.ZK_CONNECTION) : connect;
- this.basePath = config.getString(ExecConstants.ZK_ROOT);
- this.serviceName = config.getString(ExecConstants.SERVICE_NAME);
- RetryPolicy rp = new RetryNTimes(config.getInt(ExecConstants.ZK_RETRY_TIMES),
- config.getInt(ExecConstants.ZK_RETRY_DELAY));
- curator = CuratorFrameworkFactory.builder()
- .connectionTimeoutMs(config.getInt(ExecConstants.ZK_TIMEOUT))
- .retryPolicy(rp)
- .connectString(connect)
- .build();
- curator.getConnectionStateListenable().addListener(new InitialConnectionListener());
- discovery = getDiscovery();
- serviceCache = discovery.
- serviceCacheBuilder()
- .name(serviceName)
- .refreshPaddingMs(config.getInt(ExecConstants.ZK_REFRESH))
- .build();
- }
-
- public void start(long millisToWait) throws Exception {
- logger.debug("Starting ZKClusterCoordination.");
- curator.start();
- discovery.start();
- serviceCache.start();
- serviceCache.addListener(new ZKListener());
-
- if(millisToWait != 0){
- boolean success = this.initialConnection.await(millisToWait, TimeUnit.MILLISECONDS);
- if(!success) throw new IOException(String.format("Failure to connect to the zookeeper cluster service within the allotted time of %d milliseconds.", millisToWait));
- }else{
- this.initialConnection.await();
- }
-
-
- updateEndpoints();
- }
-
- private class InitialConnectionListener implements ConnectionStateListener{
-
- @Override
- public void stateChanged(CuratorFramework client, ConnectionState newState) {
- if(newState == ConnectionState.CONNECTED){
- ZKClusterCoordinator.this.initialConnection.countDown();
- client.getConnectionStateListenable().removeListener(this);
- }
- }
-
- }
-
- private class ZKListener implements ServiceCacheListener {
-
- @Override
- public void stateChanged(CuratorFramework client, ConnectionState newState) {
- }
-
- @Override
- public void cacheChanged() {
- logger.debug("Cache changed, updating.");
- updateEndpoints();
- }
- }
-
- public void close() throws IOException {
- serviceCache.close();
- discovery.close();
- curator.close();
- }
-
- @Override
- public RegistrationHandle register(DrillbitEndpoint data) {
- try {
- ServiceInstance<DrillbitEndpoint> serviceInstance = getServiceInstance(data);
- discovery.registerService(serviceInstance);
- return new ZKRegistrationHandle(serviceInstance.getId());
- } catch (Exception e) {
- throw propagate(e);
- }
- }
-
- @Override
- public void unregister(RegistrationHandle handle) {
- if (!(handle instanceof ZKRegistrationHandle)) throw new UnsupportedOperationException("Unknown handle type");
-
- ZKRegistrationHandle h = (ZKRegistrationHandle) handle;
- try {
- ServiceInstance<DrillbitEndpoint> serviceInstance = ServiceInstance.<DrillbitEndpoint>builder()
- .address("")
- .port(0)
- .id(h.id)
- .name(serviceName)
- .build();
- discovery.unregisterService(serviceInstance);
- } catch (Exception e) {
- propagate(e);
- }
- }
-
- @Override
- public Collection<DrillbitEndpoint> getAvailableEndpoints() {
- return this.endpoints;
- }
-
- private void updateEndpoints() {
- try {
- endpoints = transform(discovery.queryForInstances(serviceName),
- new Function<ServiceInstance<DrillbitEndpoint>, DrillbitEndpoint>() {
- @Override
- public DrillbitEndpoint apply(ServiceInstance<DrillbitEndpoint> input) {
- return input.getPayload();
- }
- });
- } catch (Exception e) {
- logger.error("Failure while update Drillbit service location cache.", e);
- }
- }
-
- private ServiceInstance<DrillbitEndpoint> getServiceInstance(DrillbitEndpoint endpoint) throws Exception {
- return ServiceInstance.<DrillbitEndpoint>builder()
- .name(serviceName)
- .payload(endpoint)
- .build();
- }
-
-
- public ServiceDiscovery<DrillbitEndpoint> getDiscovery() {
- return ServiceDiscoveryBuilder
- .builder(DrillbitEndpoint.class)
- .basePath(basePath)
- .client(curator)
- .serializer(DrillServiceInstanceHelper.SERIALIZER)
- .build();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/coord/ZKRegistrationHandle.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/ZKRegistrationHandle.java b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/ZKRegistrationHandle.java
deleted file mode 100644
index 9580d5d..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/ZKRegistrationHandle.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/**
- * 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.coord;
-
-import org.apache.drill.exec.coord.ClusterCoordinator.RegistrationHandle;
-
-public class ZKRegistrationHandle implements RegistrationHandle {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ZKRegistrationHandle.class);
-
- public final String id;
-
- public ZKRegistrationHandle(String id) {
- super();
- this.id = id;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/coord/local/LocalClusterCoordinator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/local/LocalClusterCoordinator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/local/LocalClusterCoordinator.java
new file mode 100644
index 0000000..2ddf280
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/local/LocalClusterCoordinator.java
@@ -0,0 +1,139 @@
+/**
+ * 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.coord.local;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.drill.exec.coord.ClusterCoordinator;
+import org.apache.drill.exec.coord.DistributedSemaphore;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+
+import com.google.common.collect.Maps;
+
+public class LocalClusterCoordinator extends ClusterCoordinator{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(LocalClusterCoordinator.class);
+
+ private volatile Map<RegistrationHandle, DrillbitEndpoint> endpoints = Maps.newConcurrentMap();
+ private volatile ConcurrentMap<String, DistributedSemaphore> semaphores = Maps.newConcurrentMap();
+
+ @Override
+ public void close() throws IOException {
+ endpoints.clear();
+ }
+
+ @Override
+ public void start(long millis) throws Exception {
+ logger.debug("Local Cluster Coordinator started.");
+ }
+
+ @Override
+ public RegistrationHandle register(DrillbitEndpoint data) {
+ logger.debug("Endpoint registered {}.", data);
+ Handle h = new Handle();
+ endpoints.put(h, data);
+ return h;
+ }
+
+ @Override
+ public void unregister(RegistrationHandle handle) {
+ if(handle == null) return;
+
+ endpoints.remove(handle);
+ }
+
+ @Override
+ public Collection<DrillbitEndpoint> getAvailableEndpoints() {
+ return endpoints.values();
+ }
+
+
+ private class Handle implements RegistrationHandle{
+ UUID id = UUID.randomUUID();
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + getOuterType().hashCode();
+ result = prime * result + ((id == null) ? 0 : id.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) return true;
+ if (obj == null) return false;
+ if (getClass() != obj.getClass()) return false;
+ Handle other = (Handle) obj;
+ if (!getOuterType().equals(other.getOuterType())) return false;
+ if (id == null) {
+ if (other.id != null) return false;
+ } else if (!id.equals(other.id)) return false;
+ return true;
+ }
+
+ private LocalClusterCoordinator getOuterType() {
+ return LocalClusterCoordinator.this;
+ }
+
+ }
+
+
+ @Override
+ public DistributedSemaphore getSemaphore(String name, int maximumLeases) {
+ semaphores.putIfAbsent(name, new LocalSemaphore(maximumLeases));
+ return semaphores.get(name);
+ }
+
+ public class LocalSemaphore implements DistributedSemaphore{
+
+ private final Semaphore inner;
+ private final LocalLease lease = new LocalLease();
+
+ public LocalSemaphore(int size){
+ inner = new Semaphore(size);
+ }
+
+ @Override
+ public DistributedLease acquire(long timeout, TimeUnit unit) throws Exception {
+ if(!inner.tryAcquire(timeout, unit)){
+ return null;
+ }else{
+ return lease;
+ }
+ }
+
+
+ private class LocalLease implements DistributedLease{
+
+ @Override
+ public void close() throws Exception {
+ inner.release();
+ }
+
+ }
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/coord/zk/ZKClusterCoordinator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/zk/ZKClusterCoordinator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/zk/ZKClusterCoordinator.java
new file mode 100644
index 0000000..c010728
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/zk/ZKClusterCoordinator.java
@@ -0,0 +1,208 @@
+/**
+ * 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.coord.zk;
+
+import static com.google.common.base.Throwables.propagate;
+import static com.google.common.collect.Collections2.transform;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.curator.framework.state.ConnectionStateListener;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.curator.x.discovery.ServiceCache;
+import org.apache.curator.x.discovery.ServiceDiscovery;
+import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
+import org.apache.curator.x.discovery.ServiceInstance;
+import org.apache.curator.x.discovery.details.ServiceCacheListener;
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.coord.ClusterCoordinator;
+import org.apache.drill.exec.coord.DistributedSemaphore;
+import org.apache.drill.exec.coord.DrillServiceInstanceHelper;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+
+import com.google.common.base.Function;
+
+/**
+ * Manages cluster coordination utilizing zookeeper. *
+ */
+public class ZKClusterCoordinator extends ClusterCoordinator {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ZKClusterCoordinator.class);
+
+ private CuratorFramework curator;
+ private ServiceDiscovery<DrillbitEndpoint> discovery;
+ private ServiceCache<DrillbitEndpoint> serviceCache;
+ private volatile Collection<DrillbitEndpoint> endpoints = Collections.emptyList();
+ private final String serviceName;
+ private final CountDownLatch initialConnection = new CountDownLatch(1);
+
+
+ public ZKClusterCoordinator(DrillConfig config) throws IOException{
+ this(config, null);
+ }
+
+ public ZKClusterCoordinator(DrillConfig config, String connect) throws IOException {
+ connect = connect == null || connect.isEmpty() ? config.getString(ExecConstants.ZK_CONNECTION) : connect;
+ this.serviceName = config.getString(ExecConstants.SERVICE_NAME);
+ RetryPolicy rp = new RetryNTimes(config.getInt(ExecConstants.ZK_RETRY_TIMES),
+ config.getInt(ExecConstants.ZK_RETRY_DELAY));
+ curator = CuratorFrameworkFactory.builder()
+ .namespace(config.getString(ExecConstants.ZK_ROOT))
+ .connectionTimeoutMs(config.getInt(ExecConstants.ZK_TIMEOUT))
+ .retryPolicy(rp)
+ .connectString(connect)
+ .build();
+ curator.getConnectionStateListenable().addListener(new InitialConnectionListener());
+ discovery = getDiscovery();
+ serviceCache = discovery.
+ serviceCacheBuilder()
+ .name(serviceName)
+ .build();
+ }
+
+ public CuratorFramework getCurator(){
+ return curator;
+ }
+
+ public void start(long millisToWait) throws Exception {
+ logger.debug("Starting ZKClusterCoordination.");
+ curator.start();
+ discovery.start();
+ serviceCache.start();
+ serviceCache.addListener(new ZKListener());
+
+ if(millisToWait != 0){
+ boolean success = this.initialConnection.await(millisToWait, TimeUnit.MILLISECONDS);
+ if(!success) throw new IOException(String.format("Failure to connect to the zookeeper cluster service within the allotted time of %d milliseconds.", millisToWait));
+ }else{
+ this.initialConnection.await();
+ }
+
+
+ updateEndpoints();
+ }
+
+ private class InitialConnectionListener implements ConnectionStateListener{
+
+ @Override
+ public void stateChanged(CuratorFramework client, ConnectionState newState) {
+ if(newState == ConnectionState.CONNECTED){
+ ZKClusterCoordinator.this.initialConnection.countDown();
+ client.getConnectionStateListenable().removeListener(this);
+ }
+ }
+
+ }
+
+ private class ZKListener implements ServiceCacheListener {
+
+ @Override
+ public void stateChanged(CuratorFramework client, ConnectionState newState) {
+ }
+
+ @Override
+ public void cacheChanged() {
+ logger.debug("Cache changed, updating.");
+ updateEndpoints();
+ }
+ }
+
+ public void close() throws IOException {
+ serviceCache.close();
+ discovery.close();
+ curator.close();
+ }
+
+ @Override
+ public RegistrationHandle register(DrillbitEndpoint data) {
+ try {
+ ServiceInstance<DrillbitEndpoint> serviceInstance = getServiceInstance(data);
+ discovery.registerService(serviceInstance);
+ return new ZKRegistrationHandle(serviceInstance.getId());
+ } catch (Exception e) {
+ throw propagate(e);
+ }
+ }
+
+ @Override
+ public void unregister(RegistrationHandle handle) {
+ if (!(handle instanceof ZKRegistrationHandle)) throw new UnsupportedOperationException("Unknown handle type");
+
+ ZKRegistrationHandle h = (ZKRegistrationHandle) handle;
+ try {
+ ServiceInstance<DrillbitEndpoint> serviceInstance = ServiceInstance.<DrillbitEndpoint>builder()
+ .address("")
+ .port(0)
+ .id(h.id)
+ .name(serviceName)
+ .build();
+ discovery.unregisterService(serviceInstance);
+ } catch (Exception e) {
+ propagate(e);
+ }
+ }
+
+ @Override
+ public Collection<DrillbitEndpoint> getAvailableEndpoints() {
+ return this.endpoints;
+ }
+
+
+ @Override
+ public DistributedSemaphore getSemaphore(String name, int maximumLeases) {
+ return new ZkDistributedSemaphore(curator, "/semaphore/" + name, maximumLeases);
+ }
+
+ private void updateEndpoints() {
+ try {
+ endpoints = transform(discovery.queryForInstances(serviceName),
+ new Function<ServiceInstance<DrillbitEndpoint>, DrillbitEndpoint>() {
+ @Override
+ public DrillbitEndpoint apply(ServiceInstance<DrillbitEndpoint> input) {
+ return input.getPayload();
+ }
+ });
+ } catch (Exception e) {
+ logger.error("Failure while update Drillbit service location cache.", e);
+ }
+ }
+
+ private ServiceInstance<DrillbitEndpoint> getServiceInstance(DrillbitEndpoint endpoint) throws Exception {
+ return ServiceInstance.<DrillbitEndpoint>builder()
+ .name(serviceName)
+ .payload(endpoint)
+ .build();
+ }
+
+
+ public ServiceDiscovery<DrillbitEndpoint> getDiscovery() {
+ return ServiceDiscoveryBuilder
+ .builder(DrillbitEndpoint.class)
+ .client(curator)
+ .serializer(DrillServiceInstanceHelper.SERIALIZER)
+ .build();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/coord/zk/ZKRegistrationHandle.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/zk/ZKRegistrationHandle.java b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/zk/ZKRegistrationHandle.java
new file mode 100644
index 0000000..9cbc03b
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/zk/ZKRegistrationHandle.java
@@ -0,0 +1,33 @@
+/**
+ * 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.coord.zk;
+
+import org.apache.drill.exec.coord.ClusterCoordinator;
+import org.apache.drill.exec.coord.ClusterCoordinator.RegistrationHandle;
+
+public class ZKRegistrationHandle implements RegistrationHandle {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ZKRegistrationHandle.class);
+
+ public final String id;
+
+ public ZKRegistrationHandle(String id) {
+ super();
+ this.id = id;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/coord/zk/ZkDistributedSemaphore.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/zk/ZkDistributedSemaphore.java b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/zk/ZkDistributedSemaphore.java
new file mode 100644
index 0000000..5f951d0
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/zk/ZkDistributedSemaphore.java
@@ -0,0 +1,61 @@
+/**
+ * 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.coord.zk;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2;
+import org.apache.curator.framework.recipes.locks.Lease;
+import org.apache.drill.exec.coord.DistributedSemaphore;
+import org.apache.drill.exec.coord.DistributedSemaphore.DistributedLease;
+
+public class ZkDistributedSemaphore implements DistributedSemaphore{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ZkDistributedSemaphore.class);
+
+ final InterProcessSemaphoreV2 semaphore;
+
+ public ZkDistributedSemaphore(CuratorFramework client, String path, int numberOfLeases) {
+ this.semaphore = new InterProcessSemaphoreV2(client, path, numberOfLeases);
+ }
+
+ @Override
+ public DistributedLease acquire(long time, TimeUnit unit) throws Exception {
+ Lease lease = semaphore.acquire(time, unit);
+ if(lease != null){
+ return new LeaseHolder(lease);
+ }else{
+ return null;
+ }
+ }
+
+ private class LeaseHolder implements DistributedLease{
+ Lease lease;
+
+ public LeaseHolder(Lease lease) {
+ super();
+ this.lease = lease;
+ }
+
+ @Override
+ public void close() throws Exception {
+ lease.close();
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/DotDrillFile.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/DotDrillFile.java b/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/DotDrillFile.java
new file mode 100644
index 0000000..009cd00
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/DotDrillFile.java
@@ -0,0 +1,58 @@
+/**
+ * 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.dotdrill;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.exec.store.dfs.shim.DrillFileSystem;
+import org.apache.drill.exec.store.dfs.shim.DrillInputStream;
+import org.apache.hadoop.fs.FileStatus;
+
+import com.google.common.base.Preconditions;
+
+public class DotDrillFile {
+
+ private FileStatus status;
+ private DotDrillType type;
+ private DrillFileSystem fs;
+
+ public static DotDrillFile create(DrillFileSystem fs, FileStatus status){
+ for(DotDrillType d : DotDrillType.values()){
+ if(!status.isDir() && d.matches(status)){
+ return new DotDrillFile(fs, status, d);
+ }
+ }
+ return null;
+ }
+
+ private DotDrillFile(DrillFileSystem fs, FileStatus status, DotDrillType type){
+ this.fs = fs;
+ this.status = status;
+ this.type = type;
+ }
+
+ public DotDrillType getType(){
+ return type;
+ }
+
+ public View getView(DrillConfig config) throws Exception{
+ Preconditions.checkArgument(type == DotDrillType.VIEW);
+ try(DrillInputStream is = fs.open(status.getPath())){
+ return config.getMapper().readValue(is.getInputStream(), View.class);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/DotDrillType.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/DotDrillType.java b/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/DotDrillType.java
new file mode 100644
index 0000000..74b71e7
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/DotDrillType.java
@@ -0,0 +1,52 @@
+/**
+ * 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.dotdrill;
+
+import org.apache.hadoop.fs.FileStatus;
+
+public enum DotDrillType {
+ VIEW
+ // ,FORMAT
+ // ,STATS
+ ;
+
+ private final String ending;
+
+ private DotDrillType(){
+ this.ending = '.' + name().toLowerCase() + ".drill";
+ }
+
+ public boolean matches(FileStatus status){
+ return status.getPath().getName().endsWith(ending);
+ }
+
+ public static final String DOT_DRILL_GLOB;
+
+ static{
+ StringBuffer b = new StringBuffer();
+ b.append(".{");
+ for(DotDrillType d : values()){
+ if(b.length() > 2){
+ b.append(',');
+ }
+ b.append(d.name().toLowerCase());
+ }
+ b.append("}.drill");
+ DOT_DRILL_GLOB = b.toString();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/DotDrillUtil.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/DotDrillUtil.java b/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/DotDrillUtil.java
new file mode 100644
index 0000000..ea60fae
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/DotDrillUtil.java
@@ -0,0 +1,61 @@
+/**
+ * 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.dotdrill;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.drill.exec.store.dfs.shim.DrillFileSystem;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.collect.Lists;
+
+public class DotDrillUtil {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DotDrillUtil.class);
+
+ private static List<DotDrillFile> getDrillFiles(DrillFileSystem fs, FileStatus[] statuses, DotDrillType... types){
+ List<DotDrillFile> files = Lists.newArrayList();
+ for(FileStatus s : statuses){
+ DotDrillFile f = DotDrillFile.create(fs, s);
+ if(f != null){
+ if(types.length == 0){
+ files.add(f);
+ }else{
+ for(DotDrillType t : types){
+ if(t == f.getType()){
+ files.add(f);
+ }
+ }
+ }
+
+ }
+ }
+ return files;
+ }
+
+ public static List<DotDrillFile> getDotDrills(DrillFileSystem fs, Path root, DotDrillType... types) throws IOException{
+ return getDrillFiles(fs, fs.getUnderlying().globStatus(new Path(root, "*.drill")), types);
+ }
+
+ public static List<DotDrillFile> getDotDrills(DrillFileSystem fs, Path root, String name, DotDrillType... types) throws IOException{
+ return getDrillFiles(fs, fs.getUnderlying().globStatus(new Path(root, name + DotDrillType.DOT_DRILL_GLOB)), types);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/View.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/View.java b/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/View.java
new file mode 100644
index 0000000..5a7c0f8
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/dotdrill/View.java
@@ -0,0 +1,146 @@
+/**
+ * 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.dotdrill;
+
+import java.util.List;
+
+import org.apache.drill.exec.planner.types.RelDataTypeDrillImpl;
+import org.apache.drill.exec.planner.types.RelDataTypeHolder;
+import org.eigenbase.reltype.RelDataType;
+import org.eigenbase.reltype.RelDataTypeFactory;
+import org.eigenbase.reltype.RelDataTypeField;
+import org.eigenbase.sql.type.SqlTypeName;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.google.common.collect.Lists;
+
+@JsonTypeName("view")
+public class View {
+
+ private final String name;
+ private String sql;
+ private List<FieldType> fields;
+
+ @JsonInclude(Include.NON_NULL)
+ public static class FieldType {
+ public final String name;
+ public final SqlTypeName type;
+ public final Integer precision;
+ public final Integer scale;
+
+ @JsonCreator
+ public FieldType(@JsonProperty("name") String name, @JsonProperty("type") SqlTypeName type, @JsonProperty("precision") Integer precision, @JsonProperty("scale") Integer scale){
+ this.name = name;
+ this.type = type;
+ this.precision = precision;
+ this.scale = scale;
+ }
+
+ public FieldType(String name, RelDataType dataType){
+ this.name = name;
+ this.type = dataType.getSqlTypeName();
+ Integer p = null;
+ Integer s = null;
+
+ switch(dataType.getSqlTypeName()){
+ case CHAR:
+ case BINARY:
+ case VARBINARY:
+ case VARCHAR:
+ p = dataType.getPrecision();
+ break;
+ case DECIMAL:
+ p = dataType.getPrecision();
+ s = dataType.getScale();
+ break;
+ }
+
+ this.precision = p;
+ this.scale = s;
+ }
+ }
+
+ public View(String name, String sql, RelDataType rowType){
+ this.name = name;
+ this.sql = sql;
+ fields = Lists.newArrayList();
+ for(RelDataTypeField f : rowType.getFieldList()){
+ fields.add(new FieldType(f.getName(), f.getType()));
+ }
+ }
+
+ @JsonCreator
+ public View(@JsonProperty("name") String name, @JsonProperty("sql") String sql, @JsonProperty("fields") List<FieldType> fields){
+ this.name = name;
+ this.sql = sql;
+ this.fields = fields;
+ }
+
+ public RelDataType getRowType(RelDataTypeFactory factory){
+
+ // if there are no fields defined, this is a dynamic view.
+ if(isDynamic()){
+ return new RelDataTypeDrillImpl(new RelDataTypeHolder(), factory);
+ }
+
+ List<RelDataType> types = Lists.newArrayList();
+ List<String> names = Lists.newArrayList();
+
+ for(FieldType field : fields){
+ names.add(field.name);
+ if(field.precision == null && field.scale == null){
+ types.add(factory.createSqlType(field.type));
+ }else if(field.precision != null){
+ types.add(factory.createSqlType(field.type, field.precision));
+ }else{
+ types.add(factory.createSqlType(field.type, field.precision, field.scale));
+ }
+ }
+ return factory.createStructType(types, names);
+ }
+
+ @JsonIgnore
+ public boolean isDynamic(){
+ return fields.isEmpty();
+ }
+
+ public String getSql() {
+ return sql;
+ }
+
+ public void setSql(String sql) {
+ this.sql = sql;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public List<FieldType> getFields() {
+ return fields;
+ }
+
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/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 36def17..aa35e65 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,7 +23,6 @@ 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;
@@ -35,9 +34,8 @@ 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;
+import org.apache.drill.exec.store.sys.TableProvider;
public class QueryContext{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(QueryContext.class);
@@ -57,7 +55,11 @@ public class QueryContext{
this.session = session;
this.timer = new Multitimer<>(QuerySetup.class);
this.plannerSettings = new PlannerSettings(session.getOptions());
- this.plannerSettings.setNumEndPoints(this.getActiveEndpoints().size());
+ this.plannerSettings.setNumEndPoints(this.getActiveEndpoints().size());
+ }
+
+ public TableProvider getSystemTableProvider(){
+ return drillbitContext.getSystemTableProvider();
}
public PlannerSettings getPlannerSettings(){
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
index a0c439e..4c0d3e0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
@@ -34,9 +34,10 @@ import org.apache.drill.common.types.Types;
import org.apache.drill.exec.cache.CachedVectorContainer;
import org.apache.drill.exec.cache.Counter;
import org.apache.drill.exec.cache.DistributedCache;
+import org.apache.drill.exec.cache.DistributedCache.CacheConfig;
+import org.apache.drill.exec.cache.DistributedCache.SerializationMode;
import org.apache.drill.exec.cache.DistributedMap;
import org.apache.drill.exec.cache.DistributedMultiMap;
-import org.apache.drill.exec.cache.VectorAccessibleSerializable;
import org.apache.drill.exec.compile.sig.MappingSet;
import org.apache.drill.exec.exception.ClassTransformationException;
import org.apache.drill.exec.exception.SchemaChangeException;
@@ -48,7 +49,6 @@ import org.apache.drill.exec.expr.TypeHelper;
import org.apache.drill.exec.expr.ValueVectorReadExpression;
import org.apache.drill.exec.expr.ValueVectorWriteExpression;
import org.apache.drill.exec.expr.fn.FunctionGenerationHelper;
-import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.memory.OutOfMemoryException;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.config.OrderedPartitionSender;
@@ -90,6 +90,17 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
private static final long ALLOCATOR_INITIAL_RESERVATION = 1*1024*1024;
private static final long ALLOCATOR_MAX_RESERVATION = 20L*1000*1000*1000;
+ public static final CacheConfig<String, CachedVectorContainer> SINGLE_CACHE_CONFIG = CacheConfig //
+ .newBuilder(CachedVectorContainer.class) //
+ .name("SINGLE-" + CachedVectorContainer.class.getSimpleName()) //
+ .mode(SerializationMode.DRILL_SERIALIZIABLE) //
+ .build();
+ public static final CacheConfig<String, CachedVectorContainer> MULTI_CACHE_CONFIG = CacheConfig //
+ .newBuilder(CachedVectorContainer.class) //
+ .name("MULTI-" + CachedVectorContainer.class.getSimpleName()) //
+ .mode(SerializationMode.DRILL_SERIALIZIABLE) //
+ .build();
+
public final MappingSet mainMapping = new MappingSet( (String) null, null, ClassGenerator.DEFAULT_CONSTANT_MAP,
ClassGenerator.DEFAULT_SCALAR_MAP);
public final MappingSet incomingMapping = new MappingSet("inIndex", null, "incoming", null,
@@ -115,9 +126,9 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
private int recordCount;
private final IntVector partitionKeyVector;
- private final DistributedMap<CachedVectorContainer> tableMap;
+ private final DistributedMap<String, CachedVectorContainer> tableMap;
private final Counter minorFragmentSampleCount;
- private final DistributedMultiMap<CachedVectorContainer> mmap;
+ private final DistributedMultiMap<String, CachedVectorContainer> mmap;
private final String mapKey;
private List<VectorContainer> sampledIncomingBatches;
@@ -131,8 +142,8 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
this.completionFactor = pop.getCompletionFactor();
DistributedCache cache = context.getDrillbitContext().getCache();
- this.mmap = cache.getMultiMap(CachedVectorContainer.class);
- this.tableMap = cache.getMap(CachedVectorContainer.class);
+ this.mmap = cache.getMultiMap(MULTI_CACHE_CONFIG);
+ this.tableMap = cache.getMap(SINGLE_CACHE_CONFIG);
Preconditions.checkNotNull(tableMap);
this.mapKey = String.format("%s_%d", context.getHandle().getQueryId(), context.getHandle().getMajorFragmentId());
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/BatchGroup.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/BatchGroup.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/BatchGroup.java
index a546852..2be6871 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/BatchGroup.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/BatchGroup.java
@@ -103,7 +103,7 @@ public class BatchGroup implements VectorAccessible {
watch.start();
vas.readFromStream(inputStream);
VectorContainer c = (VectorContainer) vas.get();
- logger.debug("Took {} us to read {} records", watch.elapsed(TimeUnit.MICROSECONDS), c.getRecordCount());
+// logger.debug("Took {} us to read {} records", watch.elapsed(TimeUnit.MICROSECONDS), c.getRecordCount());
spilledBatches--;
return c;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillViewTable.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillViewTable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillViewTable.java
index 6415869..453aa15 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillViewTable.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillViewTable.java
@@ -17,50 +17,53 @@
*/
package org.apache.drill.exec.planner.logical;
-import com.google.common.collect.ImmutableList;
-import net.hydromatic.optiq.impl.ViewTable;
-import org.apache.drill.exec.planner.types.DrillFixedRelDataTypeImpl;
-import org.apache.drill.exec.planner.types.RelDataTypeDrillImpl;
-import org.apache.drill.exec.planner.types.RelDataTypeHolder;
+import net.hydromatic.optiq.Schema.TableType;
+import net.hydromatic.optiq.Statistic;
+import net.hydromatic.optiq.Statistics;
+import net.hydromatic.optiq.TranslatableTable;
+
+import org.apache.drill.exec.dotdrill.View;
import org.eigenbase.rel.RelNode;
import org.eigenbase.relopt.RelOptTable;
import org.eigenbase.relopt.RelOptTable.ToRelContext;
import org.eigenbase.relopt.RelOptUtil;
-import org.eigenbase.reltype.*;
-
-import java.util.List;
+import org.eigenbase.reltype.RelDataType;
+import org.eigenbase.reltype.RelDataTypeFactory;
-public class DrillViewTable extends ViewTable {
- private RelDataType rowType;
- private boolean starSchema = false; // is the view schema "*"?
- private RelDataTypeHolder holder = new RelDataTypeHolder();
+public class DrillViewTable implements TranslatableTable{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillViewTable.class);
- public DrillViewTable(String viewSql, List<String> schemaPath, RelDataType rowType) {
- super(Object.class, null, viewSql, schemaPath);
+ private View view;
- this.rowType = rowType;
- if (rowType.getFieldCount() == 1 && rowType.getFieldNames().get(0).equals("*"))
- starSchema = true;
+ public DrillViewTable(View view){
+ this.view = view;
}
@Override
public RelDataType getRowType(RelDataTypeFactory typeFactory) {
- // if the view's schema is a "*" schema, create dynamic row type. Otherwise create fixed row type.
- if (starSchema)
- return new RelDataTypeDrillImpl(holder, typeFactory);
+ return view.getRowType(typeFactory);
+ }
- return rowType;
+ @Override
+ public Statistic getStatistic() {
+ return Statistics.UNKNOWN;
}
@Override
public RelNode toRel(ToRelContext context, RelOptTable relOptTable) {
RelDataType rowType = relOptTable.getRowType();
- RelNode rel = context.expandView(rowType, getViewSql(), getSchemaPath());
+ RelNode rel = context.expandView(rowType, view.getSql(), relOptTable.getQualifiedName());
- // if the View's field list is not "*", try to create a cast.
- if (!starSchema)
+ if (view.isDynamic()){
+ return rel;
+ }else{
+ // if the View's field list is not "*", try to create a cast.
return RelOptUtil.createCastRel(rel, rowType, true);
+ }
+ }
- return rel;
+ @Override
+ public TableType getJdbcTableType() {
+ return TableType.VIEW;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/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 d296eb1..5e25659 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,14 +26,13 @@ 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 extends JacksonSerializable implements Iterable<Map.Entry<String, StoragePluginConfig>>{
+public class StoragePlugins implements Iterable<Map.Entry<String, StoragePluginConfig>>{
private Map<String, StoragePluginConfig> storage;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/ExpandingConcurrentMap.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/ExpandingConcurrentMap.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/ExpandingConcurrentMap.java
index e348bc3..9caa29b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/ExpandingConcurrentMap.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/ExpandingConcurrentMap.java
@@ -26,7 +26,7 @@ import java.util.concurrent.ConcurrentMap;
import com.google.common.collect.Maps;
/**
- * A special type of concurrent map which attempts to create an object before returning that it does not exist. It will also provide the same functionality on it's keyset.
+ * A special type of concurrent map which attempts to create an object before returning that it does not exist. It will also provide the same functionality on it's keyset.
* @param <KEY> The key in the map.
* @param <VALUE> The value in the map.
*/
@@ -36,7 +36,7 @@ public class ExpandingConcurrentMap<KEY, VALUE> implements ConcurrentMap<KEY, VA
private final ConcurrentMap<KEY, VALUE> internalMap = Maps.newConcurrentMap();
private final DelegatingKeySet keySet = new DelegatingKeySet();
private final MapValueFactory<KEY, VALUE> fac;
-
+
/**
* Create a new ExpandingConcurrentMap.
* @param fac The object factory responsible for attempting to generate object instances.
@@ -56,10 +56,17 @@ public class ExpandingConcurrentMap<KEY, VALUE> implements ConcurrentMap<KEY, VA
return internalMap.isEmpty();
}
+ public boolean alreadyContainsKey(Object k){
+ @SuppressWarnings("unchecked") KEY key = (KEY) k;
+
+ if(internalMap.containsKey(key)) return true;
+ return false;
+ }
+
@Override
public boolean containsKey(Object k) {
@SuppressWarnings("unchecked") KEY key = (KEY) k;
-
+
if(internalMap.containsKey(key)) return true;
VALUE v = getNewEntry(k);
return v != null;
@@ -87,7 +94,7 @@ public class ExpandingConcurrentMap<KEY, VALUE> implements ConcurrentMap<KEY, VA
fac.destroy(v);
return old;
}
-
+
@Override
public VALUE put(KEY key, VALUE value) {
throw new UnsupportedOperationException();
@@ -142,7 +149,7 @@ public class ExpandingConcurrentMap<KEY, VALUE> implements ConcurrentMap<KEY, VA
public VALUE replace(KEY key, VALUE value) {
return null;
}
-
+
private class DelegatingKeySet implements Set<KEY>{
@Override
@@ -212,11 +219,11 @@ public class ExpandingConcurrentMap<KEY, VALUE> implements ConcurrentMap<KEY, VA
public void clear() {
throw new UnsupportedOperationException();
}
-
+
}
-
+
public interface MapValueFactory<KEY, VALUE> {
-
+
public VALUE create(KEY key);
public void destroy(VALUE value);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ViewHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ViewHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ViewHandler.java
index 1ca384b..ced07e5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ViewHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ViewHandler.java
@@ -21,11 +21,13 @@ import java.io.IOException;
import java.util.List;
import net.hydromatic.optiq.SchemaPlus;
+import net.hydromatic.optiq.Table;
import net.hydromatic.optiq.impl.ViewTable;
import net.hydromatic.optiq.tools.Planner;
import net.hydromatic.optiq.tools.RelConversionException;
import net.hydromatic.optiq.tools.ValidationException;
+import org.apache.drill.exec.dotdrill.View;
import org.apache.drill.exec.ops.QueryContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.planner.logical.DrillViewTable;
@@ -34,6 +36,7 @@ import org.apache.drill.exec.planner.sql.parser.SqlCreateView;
import org.apache.drill.exec.planner.sql.parser.SqlDropView;
import org.apache.drill.exec.planner.types.DrillFixedRelDataTypeImpl;
import org.apache.drill.exec.store.AbstractSchema;
+import org.apache.drill.exec.store.dfs.WorkspaceSchemaFactory.WorkspaceSchema;
import org.eigenbase.rel.RelNode;
import org.eigenbase.reltype.RelDataType;
import org.eigenbase.sql.SqlNode;
@@ -94,10 +97,14 @@ public abstract class ViewHandler extends AbstractSqlHandler{
queryRowType = new DrillFixedRelDataTypeImpl(planner.getTypeFactory(), viewFieldNames);
}
- ViewTable viewTable = new DrillViewTable(viewSql, drillSchema.getSchemaPath(), queryRowType);
+ View view = new View(createView.getName(), viewSql, queryRowType);
- boolean replaced = context.getSession().getViewStore().addView(
- schemaPath, createView.getName(), viewTable, createView.getReplace());
+ boolean replaced;
+ if (drillSchema instanceof WorkspaceSchema) {
+ replaced = ((WorkspaceSchema) drillSchema).createView(view);
+ }else{
+ return DirectPlan.createDirectPlan(context, false, "Schema provided was not a workspace schema.");
+ }
String summary = String.format("View '%s' %s successfully in '%s' schema",
createView.getName(), replaced ? "replaced" : "created", schemaPath);
@@ -129,7 +136,11 @@ public abstract class ViewHandler extends AbstractSqlHandler{
return DirectPlan.createDirectPlan(context, false, String.format("Schema '%s' is not a mutable schema. " +
"Views don't exist in this schema", schemaPath));
- context.getSession().getViewStore().dropView(schemaPath, dropView.getName());
+ if (drillSchema instanceof WorkspaceSchema) {
+ ((WorkspaceSchema) drillSchema).dropView(dropView.getName());;
+ }else{
+ return DirectPlan.createDirectPlan(context, false, "Schema provided was not a workspace schema.");
+ }
return DirectPlan.createDirectPlan(context, true,
String.format("View '%s' deleted successfully from '%s' schema", dropView.getName(), schemaPath));
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java
index 14ade39..480b34b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java
@@ -95,7 +95,7 @@ public class WritableBatch {
}
SelectionVectorMode svMode;
- if (def.hasIsSelectionVector2() && def.getIsSelectionVector2()) {
+ if (def.hasCarriesTwoByteSelectionVector() && def.getCarriesTwoByteSelectionVector()) {
svMode = SelectionVectorMode.TWO_BYTE;
} else {
svMode = SelectionVectorMode.NONE;
@@ -145,7 +145,7 @@ public class WritableBatch {
}
RecordBatchDef batchDef = RecordBatchDef.newBuilder().addAllField(metadata).setRecordCount(recordCount)
- .setIsSelectionVector2(isSV2).build();
+ .setCarriesTwoByteSelectionVector(isSV2).build();
WritableBatch b = new WritableBatch(batchDef, buffers);
return b;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/WorkEventBus.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/WorkEventBus.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/WorkEventBus.java
index 5592e35..a05041f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/WorkEventBus.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/WorkEventBus.java
@@ -17,9 +17,11 @@
*/
package org.apache.drill.exec.rpc.control;
+import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import org.apache.drill.exec.cache.DistributedMap;
import org.apache.drill.exec.exception.FragmentSetupException;
import org.apache.drill.exec.proto.BitControl.FragmentStatus;
import org.apache.drill.exec.proto.BitControl.PlanFragment;
@@ -27,6 +29,7 @@ import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
import org.apache.drill.exec.proto.UserBitShared.QueryId;
import org.apache.drill.exec.rpc.RpcException;
import org.apache.drill.exec.work.WorkManager.WorkerBee;
+import org.apache.drill.exec.work.foreman.Foreman;
import org.apache.drill.exec.work.foreman.FragmentStatusListener;
import org.apache.drill.exec.work.fragment.FragmentManager;
import org.apache.drill.exec.work.fragment.NonRootFragmentManager;
@@ -81,12 +84,16 @@ public class WorkEventBus {
public FragmentManager getFragmentManager(FragmentHandle handle){
return managers.get(handle);
}
-
+
public FragmentManager getOrCreateFragmentManager(FragmentHandle handle) throws FragmentSetupException{
FragmentManager manager = managers.get(handle);
if (manager != null) return manager;
-
- PlanFragment fragment = bee.getContext().getCache().getFragment(handle);
+ DistributedMap<FragmentHandle, PlanFragment> planCache = bee.getContext().getCache().getMap(Foreman.FRAGMENT_CACHE);
+ for(Map.Entry<FragmentHandle, PlanFragment> e : planCache.getLocalEntries()){
+ logger.debug("Key: {}", e.getKey());
+ logger.debug("Value: {}", e.getValue());
+ }
+ PlanFragment fragment = bee.getContext().getCache().getMap(Foreman.FRAGMENT_CACHE).get(handle);
if (fragment == null) {
throw new FragmentSetupException("Received batch where fragment was not in cache.");
@@ -106,7 +113,7 @@ public class WorkEventBus {
return manager;
}
-
+
public void removeFragmentManager(FragmentHandle handle){
managers.remove(handle);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/872d0abf/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 1607179..6ecffaf 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
@@ -41,13 +41,11 @@ public class UserSession {
private UserCredentials credentials;
private Map<String, String> properties;
private OptionManager options;
- private ViewStore viewStore;
public UserSession(OptionManager systemOptions, UserCredentials credentials, UserProperties properties) throws IOException{
this.credentials = credentials;
this.options = new SessionOptionManager(systemOptions);
this.properties = Maps.newHashMap();
- this.viewStore = new ViewStore();
if (properties == null) return;
for (int i=0; i<properties.getPropertiesCount(); i++) {
@@ -64,7 +62,6 @@ public class UserSession {
return user;
}
- public ViewStore getViewStore() { return viewStore; }
/**
* Update the schema path for the session.