You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by js...@apache.org on 2015/09/02 23:52:52 UTC
[1/5] drill git commit: DRILL-3598: use a factory to create the root
allocator. - made the constructor for TopLevelAllocator package private to
enforce this
Repository: drill
Updated Branches:
refs/heads/master e39fa835d -> 4b8e85ad6
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/trace/TestTraceOutputDump.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/trace/TestTraceOutputDump.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/trace/TestTraceOutputDump.java
index 1cb72ff..44f18af 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/trace/TestTraceOutputDump.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/trace/TestTraceOutputDump.java
@@ -28,7 +28,7 @@ import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.cache.VectorAccessibleSerializable;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -68,75 +68,68 @@ import com.google.common.io.Files;
* known value.
*/
public class TestTraceOutputDump extends ExecTest {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestTraceOutputDump.class);
- DrillConfig c = DrillConfig.create();
-
-
- @Test
- public void testFilter(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable
- {
-
- new NonStrictExpectations(){{
- bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
- bitContext.getConfig(); result = c;
- bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
- bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
- }};
-
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/trace/simple_trace.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
-
- while(exec.next()){
- }
-
- exec.close();
-
- if(context.getFailureCause() != null){
- throw context.getFailureCause();
- }
- assertTrue(!context.isFailed());
-
- FragmentHandle handle = context.getHandle();
-
- /* Form the file name to which the trace output will dump the record batches */
- String qid = QueryIdHelper.getQueryId(handle.getQueryId());
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestTraceOutputDump.class);
+ private final DrillConfig c = DrillConfig.create();
+
+ @Test
+ public void testFilter(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable {
+ new NonStrictExpectations() {{
+ bitContext.getMetrics(); result = new MetricRegistry();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
+ bitContext.getConfig(); result = c;
+ bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
+ bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
+ }};
+
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/trace/simple_trace.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+
+ while(exec.next()) {
+ }
- int majorFragmentId = handle.getMajorFragmentId();
- int minorFragmentId = handle.getMinorFragmentId();
+ exec.close();
- String logLocation = c.getString(ExecConstants.TRACE_DUMP_DIRECTORY);
+ if(context.getFailureCause() != null) {
+ throw context.getFailureCause();
+ }
+ assertTrue(!context.isFailed());
- System.out.println("Found log location: " + logLocation);
+ final FragmentHandle handle = context.getHandle();
- String filename = String.format("%s//%s_%d_%d_mock-scan", logLocation, qid, majorFragmentId, minorFragmentId);
+ /* Form the file name to which the trace output will dump the record batches */
+ final String qid = QueryIdHelper.getQueryId(handle.getQueryId());
+ final int majorFragmentId = handle.getMajorFragmentId();
+ final int minorFragmentId = handle.getMinorFragmentId();
- System.out.println("File Name: " + filename);
+ final String logLocation = c.getString(ExecConstants.TRACE_DUMP_DIRECTORY);
+ System.out.println("Found log location: " + logLocation);
- Configuration conf = new Configuration();
- conf.set(FileSystem.FS_DEFAULT_NAME_KEY, c.getString(ExecConstants.TRACE_DUMP_FILESYSTEM));
+ final String filename = String.format("%s//%s_%d_%d_mock-scan", logLocation, qid, majorFragmentId, minorFragmentId);
+ System.out.println("File Name: " + filename);
- FileSystem fs = FileSystem.get(conf);
- Path path = new Path(filename);
- assertTrue("Trace file does not exist", fs.exists(path));
- FSDataInputStream in = fs.open(path);
+ final Configuration conf = new Configuration();
+ conf.set(FileSystem.FS_DEFAULT_NAME_KEY, c.getString(ExecConstants.TRACE_DUMP_FILESYSTEM));
- VectorAccessibleSerializable wrap = new VectorAccessibleSerializable(context.getAllocator());
- wrap.readFromStream(in);
- VectorAccessible container = wrap.get();
+ final FileSystem fs = FileSystem.get(conf);
+ final Path path = new Path(filename);
+ assertTrue("Trace file does not exist", fs.exists(path));
+ final FSDataInputStream in = fs.open(path);
- /* Assert there are no selection vectors */
- assertTrue(wrap.getSv2() == null);
+ final VectorAccessibleSerializable wrap = new VectorAccessibleSerializable(context.getAllocator());
+ wrap.readFromStream(in);
+ final VectorAccessible container = wrap.get();
- /* Assert there is only one record */
- assertTrue(container.getRecordCount() == 1);
+ /* Assert there are no selection vectors */
+ assertTrue(wrap.getSv2() == null);
- /* Read the Integer value and ASSERT its Integer.MIN_VALUE */
- int value = (int) container.iterator().next().getValueVector().getAccessor().getObject(0);
- assertTrue(value == Integer.MIN_VALUE);
- }
+ /* Assert there is only one record */
+ assertTrue(container.getRecordCount() == 1);
+ /* Read the Integer value and ASSERT its Integer.MIN_VALUE */
+ final int value = (int) container.iterator().next().getValueVector().getAccessor().getObject(0);
+ assertTrue(value == Integer.MIN_VALUE);
+ }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/union/TestSimpleUnion.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/union/TestSimpleUnion.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/union/TestSimpleUnion.java
index 07de27f..7b0f137 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/union/TestSimpleUnion.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/union/TestSimpleUnion.java
@@ -27,7 +27,7 @@ import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -46,41 +46,35 @@ import com.google.common.base.Charsets;
import com.google.common.io.Files;
public class TestSimpleUnion extends ExecTest {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestSimpleUnion.class);
- DrillConfig c = DrillConfig.create();
-
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestSimpleUnion.class);
+ private final DrillConfig c = DrillConfig.create();
@Test
- public void testUnion(@Injectable final DrillbitContext bitContext, @Injectable UserServer.UserClientConnection connection) throws Throwable{
-
-
- new NonStrictExpectations(){{
+ public void testUnion(@Injectable final DrillbitContext bitContext, @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/union/test1.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/union/test1.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
-
- int[] counts = new int[]{100,50};
- int i=0;
- while(exec.next()){
+ final int[] counts = new int[]{100,50};
+ int i = 0;
+ while(exec.next()) {
System.out.println("iteration count:" + exec.getRecordCount());
assertEquals(counts[i++], exec.getRecordCount());
}
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
-
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestLoad.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestLoad.java b/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestLoad.java
index f57e765..c3af451 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestLoad.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestLoad.java
@@ -23,6 +23,7 @@ import io.netty.buffer.DrillBuf;
import java.util.List;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.expression.ExpressionPosition;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.types.TypeProtos;
@@ -30,7 +31,7 @@ import org.apache.drill.common.types.TypeProtos.MinorType;
import org.apache.drill.common.types.Types;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.record.MaterializedField;
import org.apache.drill.exec.record.RecordBatchLoader;
import org.apache.drill.exec.record.VectorWrapper;
@@ -45,31 +46,32 @@ import org.junit.Test;
import com.google.common.collect.Lists;
public class TestLoad extends ExecTest {
+ private final DrillConfig drillConfig = DrillConfig.create();
@Test
public void testLoadValueVector() throws Exception {
- BufferAllocator allocator = new TopLevelAllocator();
- ValueVector fixedV = new IntVector(MaterializedField.create(new SchemaPath("ints", ExpressionPosition.UNKNOWN),
+ final BufferAllocator allocator = RootAllocatorFactory.newRoot(drillConfig);
+ final ValueVector fixedV = new IntVector(MaterializedField.create(new SchemaPath("ints", ExpressionPosition.UNKNOWN),
Types.required(MinorType.INT)), allocator);
- ValueVector varlenV = new VarCharVector(MaterializedField.create(
+ final ValueVector varlenV = new VarCharVector(MaterializedField.create(
new SchemaPath("chars", ExpressionPosition.UNKNOWN), Types.required(MinorType.VARCHAR)), allocator);
- ValueVector nullableVarlenV = new NullableVarCharVector(MaterializedField.create(new SchemaPath("chars",
+ final ValueVector nullableVarlenV = new NullableVarCharVector(MaterializedField.create(new SchemaPath("chars",
ExpressionPosition.UNKNOWN), Types.optional(MinorType.VARCHAR)), allocator);
- List<ValueVector> vectors = Lists.newArrayList(fixedV, varlenV, nullableVarlenV);
- for (ValueVector v : vectors) {
+ final List<ValueVector> vectors = Lists.newArrayList(fixedV, varlenV, nullableVarlenV);
+ for (final ValueVector v : vectors) {
AllocationHelper.allocate(v, 100, 50);
v.getMutator().generateTestData(100);
}
- WritableBatch writableBatch = WritableBatch.getBatchNoHV(100, vectors, false);
- RecordBatchLoader batchLoader = new RecordBatchLoader(allocator);
- ByteBuf[] byteBufs = writableBatch.getBuffers();
+ final WritableBatch writableBatch = WritableBatch.getBatchNoHV(100, vectors, false);
+ final RecordBatchLoader batchLoader = new RecordBatchLoader(allocator);
+ final ByteBuf[] byteBufs = writableBatch.getBuffers();
int bytes = 0;
for (int i = 0; i < byteBufs.length; i++) {
bytes += byteBufs[i].writerIndex();
}
- DrillBuf byteBuf = allocator.buffer(bytes);
+ final DrillBuf byteBuf = allocator.buffer(bytes);
int index = 0;
for (int i = 0; i < byteBufs.length; i++) {
byteBufs[i].readBytes(byteBuf, index, byteBufs[i].writerIndex());
@@ -80,7 +82,7 @@ public class TestLoad extends ExecTest {
batchLoader.load(writableBatch.getDef(), byteBuf);
boolean firstColumn = true;
int recordCount = 0;
- for (VectorWrapper<?> v : batchLoader) {
+ for (final VectorWrapper<?> v : batchLoader) {
if (firstColumn) {
firstColumn = false;
} else {
@@ -96,15 +98,15 @@ public class TestLoad extends ExecTest {
for (int r = 0; r < batchLoader.getRecordCount(); r++) {
boolean first = true;
recordCount++;
- for (VectorWrapper<?> v : batchLoader) {
+ for (final VectorWrapper<?> v : batchLoader) {
if (first) {
first = false;
} else {
System.out.print("\t");
}
- ValueVector.Accessor accessor = v.getValueVector().getAccessor();
+ final ValueVector.Accessor accessor = v.getValueVector().getAccessor();
if (v.getField().getType().getMinorType() == TypeProtos.MinorType.VARCHAR) {
- Object obj = accessor.getObject(r);
+ final Object obj = accessor.getObject(r);
if (obj != null) {
System.out.print(accessor.getObject(r));
} else {
@@ -122,5 +124,4 @@ public class TestLoad extends ExecTest {
batchLoader.clear();
writableBatch.clear();
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestValueVector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestValueVector.java b/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestValueVector.java
index 4e299dc..af4f2f0 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestValueVector.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestValueVector.java
@@ -27,6 +27,7 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import io.netty.buffer.DrillBuf;
import org.apache.drill.common.AutoCloseables;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.types.TypeProtos;
import org.apache.drill.common.types.Types;
@@ -44,8 +45,9 @@ import org.apache.drill.exec.expr.holders.RepeatedVarBinaryHolder;
import org.apache.drill.exec.expr.holders.UInt1Holder;
import org.apache.drill.exec.expr.holders.UInt4Holder;
import org.apache.drill.exec.expr.holders.VarCharHolder;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.proto.UserBitShared;
+import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.record.MaterializedField;
import org.apache.drill.exec.vector.BaseValueVector;
import org.apache.drill.exec.vector.BitVector;
@@ -69,11 +71,13 @@ public class TestValueVector extends ExecTest {
private final static byte[] STR2 = new String("BBBBBBBBB2").getBytes(Charset.forName("UTF-8"));
private final static byte[] STR3 = new String("CCCC3").getBytes(Charset.forName("UTF-8"));
- private TopLevelAllocator allocator;
+ private DrillConfig drillConfig;
+ private BufferAllocator allocator;
@Before
public void init() {
- allocator = new TopLevelAllocator();
+ drillConfig = DrillConfig.create();
+ allocator = RootAllocatorFactory.newRoot(drillConfig);
}
@After
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestDrillbitResilience.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestDrillbitResilience.java b/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestDrillbitResilience.java
index ce09f68..c4a1a76 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestDrillbitResilience.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestDrillbitResilience.java
@@ -46,7 +46,7 @@ import org.apache.drill.exec.client.DrillClient;
import org.apache.drill.exec.exception.DrillbitStartupException;
import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.physical.impl.ScreenCreator;
import org.apache.drill.exec.physical.impl.SingleSenderCreator.SingleSenderRootExec;
import org.apache.drill.exec.physical.impl.mergereceiver.MergingRecordBatch;
@@ -234,7 +234,7 @@ public class TestDrillbitResilience extends DrillTest {
*/
private static void assertDrillbitsOk() {
final SingleRowListener listener = new SingleRowListener() {
- private final BufferAllocator bufferAllocator = new TopLevelAllocator(zkHelper.getConfig());
+ private final BufferAllocator bufferAllocator = RootAllocatorFactory.newRoot(zkHelper.getConfig());
private final RecordBatchLoader loader = new RecordBatchLoader(bufferAllocator);
@Override
@@ -929,7 +929,7 @@ public class TestDrillbitResilience extends DrillTest {
@Test // DRILL-3065
public void failsAfterMSorterSorting() {
final String query = "select n_name from cp.`tpch/nation.parquet` order by n_name";
- Class<? extends Exception> typeOfException = RuntimeException.class;
+ final Class<? extends Exception> typeOfException = RuntimeException.class;
final long before = countAllocatedMemory();
final String controls = Controls.newBuilder()
@@ -944,7 +944,7 @@ public class TestDrillbitResilience extends DrillTest {
@Test // DRILL-3085
public void failsAfterMSorterSetup() {
final String query = "select n_name from cp.`tpch/nation.parquet` order by n_name";
- Class<? extends Exception> typeOfException = RuntimeException.class;
+ final Class<? extends Exception> typeOfException = RuntimeException.class;
final long before = countAllocatedMemory();
final String controls = Controls.newBuilder()
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/store/TestDirectCodecFactory.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/TestDirectCodecFactory.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/TestDirectCodecFactory.java
index 644144e..3c5cbf7 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/TestDirectCodecFactory.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/TestDirectCodecFactory.java
@@ -23,9 +23,10 @@ import java.nio.ByteBuffer;
import java.util.Random;
import org.apache.drill.common.DeferredException;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.store.parquet.DirectCodecFactory;
import org.apache.drill.exec.store.parquet.DirectCodecFactory.ByteBufBytesInput;
import org.apache.drill.exec.store.parquet.DirectCodecFactory.DirectBytesDecompressor;
@@ -38,7 +39,8 @@ import parquet.hadoop.CodecFactory.BytesCompressor;
import parquet.hadoop.metadata.CompressionCodecName;
public class TestDirectCodecFactory extends ExecTest {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestDirectCodecFactory.class);
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestDirectCodecFactory.class);
+ private final DrillConfig drillConfig = DrillConfig.create();
private static enum Decompression {
ON_HEAP, OFF_HEAP, DRILLBUF
@@ -47,14 +49,14 @@ public class TestDirectCodecFactory extends ExecTest {
private void test(int size, CompressionCodecName codec, boolean useOnHeapCompression, Decompression decomp) {
DrillBuf rawBuf = null;
DrillBuf outBuf = null;
- try (BufferAllocator allocator = new TopLevelAllocator();
- DirectCodecFactory codecFactory = new DirectCodecFactory(new Configuration(), allocator)) {
+ try (final BufferAllocator allocator = RootAllocatorFactory.newRoot(drillConfig);
+ final DirectCodecFactory codecFactory = new DirectCodecFactory(new Configuration(), allocator)) {
try {
rawBuf = allocator.buffer(size);
final byte[] rawArr = new byte[size];
outBuf = allocator.buffer(size * 2);
- Random r = new Random();
- byte[] random = new byte[1024];
+ final Random r = new Random();
+ final byte[] random = new byte[1024];
int pos = 0;
while (pos < size) {
r.nextBytes(random);
@@ -63,10 +65,10 @@ public class TestDirectCodecFactory extends ExecTest {
pos += random.length;
}
- BytesCompressor c = codecFactory.getCompressor(codec, 64 * 1024);
- DirectBytesDecompressor d = codecFactory.getDecompressor(codec);
+ final BytesCompressor c = codecFactory.getCompressor(codec, 64 * 1024);
+ final DirectBytesDecompressor d = codecFactory.getDecompressor(codec);
- BytesInput compressed;
+ final BytesInput compressed;
if (useOnHeapCompression) {
compressed = c.compress(BytesInput.from(rawArr));
} else {
@@ -75,8 +77,8 @@ public class TestDirectCodecFactory extends ExecTest {
switch (decomp) {
case DRILLBUF: {
- ByteBuffer buf = compressed.toByteBuffer();
- DrillBuf b = allocator.buffer(buf.capacity());
+ final ByteBuffer buf = compressed.toByteBuffer();
+ final DrillBuf b = allocator.buffer(buf.capacity());
try {
b.writeBytes(buf);
d.decompress(b, (int) compressed.size(), outBuf, size);
@@ -90,11 +92,11 @@ public class TestDirectCodecFactory extends ExecTest {
}
case OFF_HEAP: {
- ByteBuffer buf = compressed.toByteBuffer();
- DrillBuf b = allocator.buffer(buf.capacity());
+ final ByteBuffer buf = compressed.toByteBuffer();
+ final DrillBuf b = allocator.buffer(buf.capacity());
try {
b.writeBytes(buf);
- BytesInput input = d.decompress(new ByteBufBytesInput(b), size);
+ final BytesInput input = d.decompress(new ByteBufBytesInput(b), size);
Assert.assertArrayEquals(input.toByteArray(), rawArr);
} finally {
b.release();
@@ -102,14 +104,14 @@ public class TestDirectCodecFactory extends ExecTest {
break;
}
case ON_HEAP: {
- byte[] buf = compressed.toByteArray();
- BytesInput input = d.decompress(BytesInput.from(buf), size);
+ final byte[] buf = compressed.toByteArray();
+ final BytesInput input = d.decompress(BytesInput.from(buf), size);
Assert.assertArrayEquals(input.toByteArray(), rawArr);
break;
}
}
} catch (Exception e) {
- String msg = String.format(
+ final String msg = String.format(
"Failure while testing Codec: %s, OnHeapCompressionInput: %s, Decompression Mode: %s, Data Size: %d",
codec.name(),
useOnHeapCompression, decomp.name(), size);
@@ -128,14 +130,14 @@ public class TestDirectCodecFactory extends ExecTest {
@Test
public void compressionCodecs() throws Exception {
- int[] sizes = { 4 * 1024, 1 * 1024 * 1024 };
- boolean[] comp = { true, false };
+ final int[] sizes = { 4 * 1024, 1 * 1024 * 1024 };
+ final boolean[] comp = { true, false };
- try (DeferredException ex = new DeferredException()) {
- for (int size : sizes) {
- for (boolean useOnHeapComp : comp) {
- for (Decompression decomp : Decompression.values()) {
- for (CompressionCodecName codec : CompressionCodecName.values()) {
+ try (final DeferredException ex = new DeferredException()) {
+ for (final int size : sizes) {
+ for (final boolean useOnHeapComp : comp) {
+ for (final Decompression decomp : Decompression.values()) {
+ for (final CompressionCodecName codec : CompressionCodecName.values()) {
if (codec == CompressionCodecName.LZO) {
// not installed as gpl.
continue;
@@ -151,5 +153,4 @@ public class TestDirectCodecFactory extends ExecTest {
}
}
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/ParquetRecordReaderTest.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/ParquetRecordReaderTest.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/ParquetRecordReaderTest.java
index af1b896..8ded703 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/ParquetRecordReaderTest.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/ParquetRecordReaderTest.java
@@ -41,7 +41,7 @@ import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.impl.OutputMutator;
import org.apache.drill.exec.proto.BitControl;
@@ -611,9 +611,9 @@ public class ParquetRecordReaderTest extends BaseTestQuery {
@Ignore
public void testPerformance(@Injectable final DrillbitContext bitContext,
@Injectable UserServer.UserClientConnection connection) throws Exception {
- DrillConfig c = DrillConfig.create();
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, BitControl.PlanFragment.getDefaultInstance(), connection, registry);
+ final DrillConfig c = DrillConfig.create();
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, BitControl.PlanFragment.getDefaultInstance(), connection, registry);
// new NonStrictExpectations() {
// {
@@ -642,7 +642,7 @@ public class ParquetRecordReaderTest extends BaseTestQuery {
int totalRowCount = 0;
final FileSystem fs = new CachedSingleFileSystem(fileName);
- final BufferAllocator allocator = new TopLevelAllocator();
+ final BufferAllocator allocator = RootAllocatorFactory.newRoot(c);
for(int i = 0; i < 25; i++) {
final ParquetRecordReader rr = new ParquetRecordReader(context, 256000, fileName, 0, fs,
new DirectCodecFactory(dfsConfig, allocator), f.getParquetMetadata(), columns);
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/testing/TestResourceLeak.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/testing/TestResourceLeak.java b/exec/java-exec/src/test/java/org/apache/drill/exec/testing/TestResourceLeak.java
index d7e317c..dc539c5 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/testing/TestResourceLeak.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/testing/TestResourceLeak.java
@@ -19,7 +19,9 @@ package org.apache.drill.exec.testing;
import com.google.common.base.Charsets;
import com.google.common.io.Resources;
+
import io.netty.buffer.DrillBuf;
+
import org.apache.drill.QueryTestUtil;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.exceptions.UserRemoteException;
@@ -33,7 +35,7 @@ import org.apache.drill.exec.expr.annotations.Output;
import org.apache.drill.exec.expr.annotations.Param;
import org.apache.drill.exec.expr.holders.Float8Holder;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.server.Drillbit;
import org.apache.drill.exec.server.RemoteServiceSet;
import org.apache.drill.test.DrillTest;
@@ -43,6 +45,7 @@ import org.junit.BeforeClass;
import org.junit.Test;
import javax.inject.Inject;
+
import java.io.IOException;
import java.net.URL;
import java.util.Properties;
@@ -66,7 +69,7 @@ public class TestResourceLeak extends DrillTest {
@BeforeClass
public static void openClient() throws Exception {
config = DrillConfig.create(TEST_CONFIGURATIONS);
- allocator = new TopLevelAllocator(config);
+ allocator = RootAllocatorFactory.newRoot(config);
serviceSet = RemoteServiceSet.getLocalServiceSet();
bit = new Drillbit(config, serviceSet);
@@ -74,7 +77,7 @@ public class TestResourceLeak extends DrillTest {
client = QueryTestUtil.createClient(config, serviceSet, 2, null);
}
- @Test()
+ @Test
public void tpch01() throws Exception {
final String query = getFile("memory/tpch01_memory_leak.sql");
try {
@@ -89,7 +92,7 @@ public class TestResourceLeak extends DrillTest {
}
private static String getFile(String resource) throws IOException {
- URL url = Resources.getResource(resource);
+ final URL url = Resources.getResource(resource);
if (url == null) {
throw new IOException(String.format("Unable to find path %s.", resource));
}
@@ -115,12 +118,13 @@ public class TestResourceLeak extends DrillTest {
@Inject DrillBuf buf;
@Output Float8Holder out;
+ @Override
public void setup() {}
+ @Override
public void eval() {
buf.getAllocator().buffer(1);
out.value = in.value;
}
-
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/vector/TestSplitAndTransfer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/vector/TestSplitAndTransfer.java b/exec/java-exec/src/test/java/org/apache/drill/exec/vector/TestSplitAndTransfer.java
index 4b3aa8a..a0a0898 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/vector/TestSplitAndTransfer.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/vector/TestSplitAndTransfer.java
@@ -17,10 +17,11 @@
*/
package org.apache.drill.exec.vector;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.types.TypeProtos.MinorType;
import org.apache.drill.common.types.Types;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.record.MaterializedField;
import org.apache.drill.exec.record.TransferPair;
import org.apache.drill.exec.vector.NullableVarCharVector.Accessor;
@@ -28,37 +29,37 @@ import org.junit.Assert;
import org.junit.Test;
public class TestSplitAndTransfer {
-
@Test
public void test() throws Exception {
- BufferAllocator allocator = new TopLevelAllocator();
- MaterializedField field = MaterializedField.create("field", Types.optional(MinorType.VARCHAR));
- NullableVarCharVector varCharVector = new NullableVarCharVector(field, allocator);
+ final DrillConfig drillConfig = DrillConfig.create();
+ final BufferAllocator allocator = RootAllocatorFactory.newRoot(drillConfig);
+ final MaterializedField field = MaterializedField.create("field", Types.optional(MinorType.VARCHAR));
+ final NullableVarCharVector varCharVector = new NullableVarCharVector(field, allocator);
varCharVector.allocateNew(10000, 1000);
- String[] compareArray = new String[500];
+ final String[] compareArray = new String[500];
for (int i = 0; i < 500; i += 3) {
- String s = String.format("%010d", i);
+ final String s = String.format("%010d", i);
varCharVector.getMutator().set(i, s.getBytes());
compareArray[i] = s;
}
varCharVector.getMutator().setValueCount(500);
- TransferPair tp = varCharVector.getTransferPair();
- NullableVarCharVector newVarCharVector = (NullableVarCharVector) tp.getTo();
- Accessor accessor = newVarCharVector.getAccessor();
- int[][] startLengths = {{0, 201}, {201, 200}, {401, 99}};
+ final TransferPair tp = varCharVector.getTransferPair();
+ final NullableVarCharVector newVarCharVector = (NullableVarCharVector) tp.getTo();
+ final Accessor accessor = newVarCharVector.getAccessor();
+ final int[][] startLengths = {{0, 201}, {201, 200}, {401, 99}};
for (int[] startLength : startLengths) {
- int start = startLength[0];
- int length = startLength[1];
+ final int start = startLength[0];
+ final int length = startLength[1];
tp.splitAndTransfer(start, length);
newVarCharVector.getMutator().setValueCount(length);
for (int i = 0; i < length; i++) {
- boolean expectedSet = ((start + i) % 3) == 0;
+ final boolean expectedSet = ((start + i) % 3) == 0;
if (expectedSet) {
- byte[] expectedValue = compareArray[start + i].getBytes();
+ final byte[] expectedValue = compareArray[start + i].getBytes();
Assert.assertFalse(accessor.isNull(i));
// System.out.println(new String(accessor.get(i)));
Assert.assertArrayEquals(expectedValue, accessor.get(i));
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/TestEmptyPopulation.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/TestEmptyPopulation.java b/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/TestEmptyPopulation.java
index f7f2706..5bb6815 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/TestEmptyPopulation.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/TestEmptyPopulation.java
@@ -19,9 +19,11 @@ package org.apache.drill.exec.vector.complex;
import static org.apache.drill.TestBuilder.listOf;
import static org.apache.drill.TestBuilder.mapOf;
+
import org.apache.drill.BaseTestQuery;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.vector.UInt4Vector;
import org.junit.Assert;
import org.junit.Before;
@@ -36,7 +38,8 @@ public class TestEmptyPopulation extends BaseTestQuery {
private UInt4Vector.Accessor accessor;
private UInt4Vector.Mutator mutator;
private EmptyValuePopulator populator;
- private BufferAllocator allocator = new TopLevelAllocator();
+ private final DrillConfig drillConfig = DrillConfig.create();
+ private final BufferAllocator allocator = RootAllocatorFactory.newRoot(drillConfig);
@Before
@@ -310,5 +313,4 @@ public class TestEmptyPopulation extends BaseTestQuery {
.baselineValues(listOf(listOf(listOf(2L), listOf()), listOf()))
.go();
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestRepeated.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestRepeated.java b/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestRepeated.java
index 5640c8e..023dc08 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestRepeated.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestRepeated.java
@@ -20,10 +20,11 @@ package org.apache.drill.exec.vector.complex.writer;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.expr.holders.BigIntHolder;
import org.apache.drill.exec.expr.holders.IntHolder;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.vector.complex.MapVector;
import org.apache.drill.exec.vector.complex.fn.JsonWriter;
import org.apache.drill.exec.vector.complex.impl.ComplexWriterImpl;
@@ -41,11 +42,12 @@ import com.google.common.base.Charsets;
public class TestRepeated {
// private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestRepeated.class);
+ private static final DrillConfig drillConfig = DrillConfig.create();
private static BufferAllocator allocator;
@BeforeClass
public static void setupAllocator() {
- allocator = new TopLevelAllocator();
+ allocator = RootAllocatorFactory.newRoot(drillConfig);
}
@AfterClass
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/jdbc/src/main/java/org/apache/drill/jdbc/impl/DrillConnectionImpl.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/impl/DrillConnectionImpl.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/impl/DrillConnectionImpl.java
index 25fcf26..b3fa0ea 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/impl/DrillConnectionImpl.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/impl/DrillConnectionImpl.java
@@ -39,7 +39,7 @@ import org.apache.drill.common.exceptions.DrillRuntimeException;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.client.DrillClient;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.rpc.RpcException;
import org.apache.drill.exec.server.Drillbit;
import org.apache.drill.exec.server.RemoteServiceSet;
@@ -90,7 +90,7 @@ class DrillConnectionImpl extends AvaticaConnection
}
final DrillConfig dConfig = DrillConfig.create(info);
- this.allocator = new TopLevelAllocator(dConfig);
+ this.allocator = RootAllocatorFactory.newRoot(dConfig);
RemoteServiceSet set = GlobalServiceSetReference.SETS.get();
if (set == null) {
// We're embedded; start a local drill bit.
@@ -120,12 +120,12 @@ class DrillConnectionImpl extends AvaticaConnection
this.client.connect(null, info);
} else if(config.isDirect()) {
final DrillConfig dConfig = DrillConfig.forClient();
- this.allocator = new TopLevelAllocator(dConfig);
+ this.allocator = RootAllocatorFactory.newRoot(dConfig);
this.client = new DrillClient(dConfig, true); // Get a direct connection
this.client.connect(config.getZookeeperConnectionString(), info);
} else {
final DrillConfig dConfig = DrillConfig.forClient();
- this.allocator = new TopLevelAllocator(dConfig);
+ this.allocator = RootAllocatorFactory.newRoot(dConfig);
// TODO: Check: Why does new DrillClient() create another DrillConfig,
// with enableServerConfigs true, and cause scanning for function
// implementations (needed by a server, but not by a client-only
[4/5] drill git commit: DRILL-3598: use a factory to create the root
allocator. - made the constructor for TopLevelAllocator package private to
enforce this
Posted by js...@apache.org.
DRILL-3598: use a factory to create the root allocator.
- made the constructor for TopLevelAllocator package private to enforce this
Delete a test that had been commented out for over a year, it no longer compiles due to interface changes and there is plenty of other testing for hash aggregate.
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/bfe6cfad
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/bfe6cfad
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/bfe6cfad
Branch: refs/heads/master
Commit: bfe6cfadc084d8ebba52236bbe2556d77ae357e0
Parents: e39fa83
Author: Chris Westin <cw...@yahoo.com>
Authored: Tue Aug 4 06:26:57 2015 -0700
Committer: Jason Altekruse <al...@gmail.com>
Committed: Wed Sep 2 14:05:28 2015 -0700
----------------------------------------------------------------------
.../apache/drill/exec/TestMemoryRetention.java | 54 ++--
.../apache/drill/exec/client/DrillClient.java | 33 +--
.../org/apache/drill/exec/client/DumpCat.java | 62 ++---
.../exec/client/PrintingResultsListener.java | 4 +-
.../drill/exec/memory/RootAllocatorFactory.java | 67 +++++
.../drill/exec/memory/TopLevelAllocator.java | 8 +-
.../drill/exec/server/BootStrapContext.java | 11 +-
.../java/org/apache/drill/BaseTestQuery.java | 39 +--
.../java/org/apache/drill/PlanningBase.java | 24 +-
.../apache/drill/exec/client/DumpCatTest.java | 55 ++--
.../apache/drill/exec/expr/ExpressionTest.java | 27 +-
.../fn/impl/TestByteComparisonFunctions.java | 83 +++---
.../drill/exec/fn/impl/TestMathFunctions.java | 69 +++--
.../exec/fn/impl/TestNewMathFunctions.java | 47 ++--
.../exec/fn/impl/TestRepeatedFunction.java | 44 ++--
.../apache/drill/exec/memory/TestEndianess.java | 23 +-
.../exec/physical/impl/TestCastFunctions.java | 263 +++++++++----------
.../physical/impl/TestComparisonFunctions.java | 25 +-
.../impl/TestImplicitCastFunctions.java | 37 ++-
.../exec/physical/impl/TestOptiqPlans.java | 161 ++++++------
.../physical/impl/TestSimpleFragmentRun.java | 46 ++--
.../exec/physical/impl/TestSimpleFunctions.java | 109 ++++----
.../exec/physical/impl/TestStringFunctions.java | 116 ++++----
.../drill/exec/physical/impl/agg/TestAgg.java | 63 ++---
.../physical/impl/common/TestHashTable.java | 53 ++--
.../physical/impl/filter/TestSimpleFilter.java | 51 ++--
.../exec/physical/impl/join/TestHashJoin.java | 75 +++---
.../exec/physical/impl/join/TestMergeJoin.java | 148 +++++------
.../physical/impl/limit/TestSimpleLimit.java | 56 ++--
.../impl/project/TestSimpleProjection.java | 30 +--
.../exec/physical/impl/sort/TestSimpleSort.java | 14 +-
.../impl/trace/TestTraceMultiRecordBatch.java | 58 ++--
.../impl/trace/TestTraceOutputDump.java | 113 ++++----
.../physical/impl/union/TestSimpleUnion.java | 36 ++-
.../drill/exec/record/vector/TestLoad.java | 33 +--
.../exec/record/vector/TestValueVector.java | 10 +-
.../exec/server/TestDrillbitResilience.java | 8 +-
.../exec/store/TestDirectCodecFactory.java | 51 ++--
.../store/parquet/ParquetRecordReaderTest.java | 10 +-
.../drill/exec/testing/TestResourceLeak.java | 14 +-
.../drill/exec/vector/TestSplitAndTransfer.java | 31 +--
.../vector/complex/TestEmptyPopulation.java | 8 +-
.../vector/complex/writer/TestRepeated.java | 6 +-
.../drill/jdbc/impl/DrillConnectionImpl.java | 8 +-
44 files changed, 1111 insertions(+), 1172 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/main/java/org/apache/drill/exec/TestMemoryRetention.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/TestMemoryRetention.java b/exec/java-exec/src/main/java/org/apache/drill/exec/TestMemoryRetention.java
index 37e5389..f632125 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/TestMemoryRetention.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/TestMemoryRetention.java
@@ -25,29 +25,30 @@ import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.drill.common.config.DrillConfig;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import com.google.common.collect.Lists;
public class TestMemoryRetention {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestMemoryRetention.class);
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestMemoryRetention.class);
- static final int SMALL_AVERAGE_BYTES = 1024 * 32;
- static final int LARGE_BYTES = 32 * 1024 * 1024;
- static final int PARALLEL_THREADS = 32;
- static final double SMALL_ALLOCATION_MEM = 0.20;
- static final double OVERHEAD_ALLOWANCE = 0.20;
- static final List<Integer> ALLOCATIONS;
- static final int MAX_ALLOCS = 100;
- static final AtomicInteger ALLOCS = new AtomicInteger(0);
+ private static final int SMALL_AVERAGE_BYTES = 1024 * 32;
+ private static final int LARGE_BYTES = 32 * 1024 * 1024;
+ private static final int PARALLEL_THREADS = 32;
+ private static final double SMALL_ALLOCATION_MEM = 0.20;
+ private static final double OVERHEAD_ALLOWANCE = 0.20;
+ private static final List<Integer> ALLOCATIONS;
+ private static final int MAX_ALLOCS = 100;
+ private static final AtomicInteger ALLOCS = new AtomicInteger(0);
static {
- Random r = new Random();
- long maxMemory = DrillConfig.getMaxDirectMemory();
- long maxPerThread = maxMemory / PARALLEL_THREADS;
- double smallCount = (maxPerThread * SMALL_ALLOCATION_MEM) / SMALL_AVERAGE_BYTES;
- double largeCount = (maxPerThread * (1 - SMALL_ALLOCATION_MEM - OVERHEAD_ALLOWANCE)) / LARGE_BYTES;
- List<Integer> allocations = Lists.newArrayList();
+ final Random r = new Random();
+ final long maxMemory = DrillConfig.getMaxDirectMemory();
+ final long maxPerThread = maxMemory / PARALLEL_THREADS;
+ final double smallCount = (maxPerThread * SMALL_ALLOCATION_MEM) / SMALL_AVERAGE_BYTES;
+ final double largeCount = (maxPerThread * (1 - SMALL_ALLOCATION_MEM - OVERHEAD_ALLOWANCE)) / LARGE_BYTES;
+ final List<Integer> allocations = Lists.newArrayList();
for (int i = 0; i < smallCount; i++) {
allocations.add(SMALL_AVERAGE_BYTES / 2 + r.nextInt(SMALL_AVERAGE_BYTES));
@@ -63,7 +64,7 @@ public class TestMemoryRetention {
public static void main(String[] args) throws Exception {
final DrillConfig config = DrillConfig.create();
- final TopLevelAllocator a = new TopLevelAllocator(config);
+ final BufferAllocator a = RootAllocatorFactory.newRoot(config);
for (int i = 0; i < PARALLEL_THREADS; i++) {
Alloc alloc = new Alloc(a);
alloc.start();
@@ -71,15 +72,15 @@ public class TestMemoryRetention {
}
private static class Alloc extends Thread {
- final TopLevelAllocator allocator;
+ final BufferAllocator allocator;
- Alloc(TopLevelAllocator allocator) {
+ Alloc(BufferAllocator allocator) {
this.allocator = allocator;
}
@Override
public void run() {
- Random r = new Random();
+ final Random r = new Random();
try {
if (ALLOCS.incrementAndGet() > MAX_ALLOCS) {
@@ -93,7 +94,7 @@ public class TestMemoryRetention {
logger.info("Starting alloc.");
final List<DrillBuf> bufs = Lists.newLinkedList();
- for (Integer i : ALLOCATIONS) {
+ for (final Integer i : ALLOCATIONS) {
bufs.add(allocator.buffer(i));
}
Collections.shuffle(bufs);
@@ -107,25 +108,24 @@ public class TestMemoryRetention {
} else {
d.run();
}
-
}
-
}
private static class Dealloc extends Thread {
final List<DrillBuf> bufs;
- final TopLevelAllocator a;
+ final BufferAllocator a;
- public Dealloc(List<DrillBuf> bufs, TopLevelAllocator a) {
+ public Dealloc(List<DrillBuf> bufs, BufferAllocator a) {
this.bufs = bufs;
this.a = a;
}
+ @Override
public void run() {
try {
Thread.sleep(8000);
logger.info("Starting release.");
- for (DrillBuf buf : bufs) {
+ for (final DrillBuf buf : bufs) {
buf.release();
}
logger.info("Finished release.");
@@ -139,6 +139,4 @@ public class TestMemoryRetention {
alloc.start();
}
}
-
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/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 c642c4a..911828a 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
@@ -37,7 +37,7 @@ import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.coord.ClusterCoordinator;
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.memory.RootAllocatorFactory;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
import org.apache.drill.exec.proto.UserBitShared;
@@ -123,7 +123,7 @@ public class DrillClient implements Closeable, ConnectionThrottle {
this.isDirectConnection = isDirect;
this.ownsZkConnection = coordinator == null && !isDirect;
this.ownsAllocator = allocator == null;
- this.allocator = ownsAllocator ? new TopLevelAllocator(config) : allocator;
+ this.allocator = ownsAllocator ? RootAllocatorFactory.newRoot(config) : allocator;
this.config = config;
this.clusterCoordinator = coordinator;
this.reconnectTimes = config.getInt(ExecConstants.BIT_RETRY_TIMES);
@@ -173,8 +173,8 @@ public class DrillClient implements Closeable, ConnectionThrottle {
final DrillbitEndpoint endpoint;
if (isDirectConnection) {
- String[] connectInfo = props.getProperty("drillbit").split(":");
- String port = connectInfo.length==2?connectInfo[1]:config.getString(ExecConstants.INITIAL_USER_PORT);
+ final String[] connectInfo = props.getProperty("drillbit").split(":");
+ final String port = connectInfo.length==2?connectInfo[1]:config.getString(ExecConstants.INITIAL_USER_PORT);
endpoint = DrillbitEndpoint.newBuilder()
.setAddress(connectInfo[0])
.setUserPort(Integer.parseInt(port))
@@ -189,7 +189,7 @@ public class DrillClient implements Closeable, ConnectionThrottle {
}
}
- ArrayList<DrillbitEndpoint> endpoints = new ArrayList<>(clusterCoordinator.getAvailableEndpoints());
+ final ArrayList<DrillbitEndpoint> endpoints = new ArrayList<>(clusterCoordinator.getAvailableEndpoints());
checkState(!endpoints.isEmpty(), "No DrillbitEndpoint can be found");
// shuffle the collection then get the first endpoint
Collections.shuffle(endpoints);
@@ -197,8 +197,8 @@ public class DrillClient implements Closeable, ConnectionThrottle {
}
if (props != null) {
- UserProperties.Builder upBuilder = UserProperties.newBuilder();
- for (String key : props.stringPropertyNames()) {
+ final UserProperties.Builder upBuilder = UserProperties.newBuilder();
+ for (final String key : props.stringPropertyNames()) {
upBuilder.addProperties(Property.newBuilder().setKey(key).setValue(props.getProperty(key)));
}
@@ -225,7 +225,7 @@ public class DrillClient implements Closeable, ConnectionThrottle {
retry--;
try {
Thread.sleep(this.reconnectDelay);
- ArrayList<DrillbitEndpoint> endpoints = new ArrayList<>(clusterCoordinator.getAvailableEndpoints());
+ final ArrayList<DrillbitEndpoint> endpoints = new ArrayList<>(clusterCoordinator.getAvailableEndpoints());
if (endpoints.isEmpty()) {
continue;
}
@@ -240,7 +240,7 @@ public class DrillClient implements Closeable, ConnectionThrottle {
}
private void connect(DrillbitEndpoint endpoint) throws RpcException {
- FutureHandler f = new FutureHandler();
+ final FutureHandler f = new FutureHandler();
client.connect(f, endpoint, props, getUserCredentials());
f.checkedGet();
}
@@ -285,8 +285,8 @@ public class DrillClient implements Closeable, ConnectionThrottle {
* @throws RpcException
*/
public List<QueryDataBatch> runQuery(QueryType type, String plan) throws RpcException {
- UserProtos.RunQuery query = newBuilder().setResultsMode(STREAM_FULL).setType(type).setPlan(plan).build();
- ListHoldingResultsListener listener = new ListHoldingResultsListener(query);
+ final UserProtos.RunQuery query = newBuilder().setResultsMode(STREAM_FULL).setType(type).setPlan(plan).build();
+ final ListHoldingResultsListener listener = new ListHoldingResultsListener(query);
client.submitQuery(listener, query);
return listener.getResults();
}
@@ -329,17 +329,15 @@ public class DrillClient implements Closeable, ConnectionThrottle {
* Submits a Logical plan for direct execution (bypasses parsing)
*
* @param plan the plan to execute
- * @return a handle for the query result
- * @throws RpcException
*/
public void runQuery(QueryType type, String plan, UserResultsListener resultsListener) {
client.submitQuery(resultsListener, newBuilder().setResultsMode(STREAM_FULL).setType(type).setPlan(plan).build());
}
private class ListHoldingResultsListener implements UserResultsListener {
- private Vector<QueryDataBatch> results = new Vector<>();
- private SettableFuture<List<QueryDataBatch>> future = SettableFuture.create();
- private UserProtos.RunQuery query ;
+ private final Vector<QueryDataBatch> results = new Vector<>();
+ private final SettableFuture<List<QueryDataBatch>> future = SettableFuture.create();
+ private final UserProtos.RunQuery query ;
public ListHoldingResultsListener(UserProtos.RunQuery query) {
logger.debug( "Listener created for query \"\"\"{}\"\"\"", query );
@@ -395,11 +393,9 @@ public class DrillClient implements Closeable, ConnectionThrottle {
logger.debug("Query ID arrived: {}", QueryIdHelper.getQueryId(queryId));
}
}
-
}
private class FutureHandler extends AbstractCheckedFuture<Void, RpcException> implements RpcConnectionHandler<ServerConnection>, DrillRpcFuture<Void>{
-
protected FutureHandler() {
super( SettableFuture.<Void>create());
}
@@ -427,6 +423,5 @@ public class DrillClient implements Closeable, ConnectionThrottle {
public DrillBuf getBuffer() {
return null;
}
-
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/main/java/org/apache/drill/exec/client/DumpCat.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DumpCat.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DumpCat.java
index 55d9cf3..54a61c8 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DumpCat.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DumpCat.java
@@ -21,9 +21,10 @@ import java.io.File;
import java.io.FileInputStream;
import java.util.List;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.cache.VectorAccessibleSerializable;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.record.BatchSchema;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.apache.drill.exec.record.MaterializedField;
@@ -39,20 +40,20 @@ import com.beust.jcommander.ParameterException;
import com.google.common.collect.Lists;
public class DumpCat {
-
- private final static BufferAllocator allocator = new TopLevelAllocator();
+ private final static DrillConfig drillConfig = DrillConfig.create();
+ private final static BufferAllocator allocator = RootAllocatorFactory.newRoot(drillConfig);
public static void main(String args[]) throws Exception {
- DumpCat dumpCat = new DumpCat();
+ final DumpCat dumpCat = new DumpCat();
- Options o = new Options();
+ final Options o = new Options();
JCommander jc = null;
try {
jc = new JCommander(o, args);
jc.setProgramName("./drill_dumpcat");
} catch (ParameterException e) {
System.out.println(e.getMessage());
- String[] valid = {"-f", "file"};
+ final String[] valid = {"-f", "file"};
new JCommander(o, valid).usage();
System.exit(-1);
}
@@ -62,21 +63,19 @@ public class DumpCat {
}
/*Check if dump file exists*/
- File file = new File(o.location);
+ final File file = new File(o.location);
if (!file.exists()) {
System.out.println(String.format("Trace file %s not created", o.location));
System.exit(-1);
}
- FileInputStream input = new FileInputStream(file.getAbsoluteFile());
-
- if (o.batch < 0) {
- dumpCat.doQuery(input);
- } else {
- dumpCat.doBatch(input, o.batch, o.include_headers);
+ try (final FileInputStream input = new FileInputStream(file.getAbsoluteFile())) {
+ if (o.batch < 0) {
+ dumpCat.doQuery(input);
+ } else {
+ dumpCat.doBatch(input, o.batch, o.include_headers);
+ }
}
-
- input.close();
}
/**
@@ -86,7 +85,7 @@ public class DumpCat {
@Override
public void validate(String name, String value) throws ParameterException {
try {
- int batch = Integer.parseInt(value);
+ final int batch = Integer.parseInt(value);
if (batch < 0) {
throw new ParameterException("Parameter " + name + " should be non-negative number.");
}
@@ -140,7 +139,7 @@ public class DumpCat {
@Override
public String toString() {
String avgRecSizeStr = null;
- if (this.rows>0) {
+ if (this.rows > 0) {
avgRecSizeStr = String.format("Average Record Size : %d ", this.dataSize/this.rows);
} else {
avgRecSizeStr = "Average Record Size : 0";
@@ -169,14 +168,14 @@ public class DumpCat {
int batchNum = 0;
int emptyBatchNum = 0;
BatchSchema prevSchema = null;
- List<Integer> schemaChangeIdx = Lists.newArrayList();
+ final List<Integer> schemaChangeIdx = Lists.newArrayList();
- BatchMetaInfo aggBatchMetaInfo = new BatchMetaInfo();
+ final BatchMetaInfo aggBatchMetaInfo = new BatchMetaInfo();
while (input.available() > 0) {
- VectorAccessibleSerializable vcSerializable = new VectorAccessibleSerializable(DumpCat.allocator);
+ final VectorAccessibleSerializable vcSerializable = new VectorAccessibleSerializable(DumpCat.allocator);
vcSerializable.readFromStream(input);
- VectorContainer vectorContainer = (VectorContainer) vcSerializable.get();
+ final VectorContainer vectorContainer = (VectorContainer) vcSerializable.get();
aggBatchMetaInfo.add(getBatchMetaInfo(vcSerializable));
@@ -225,7 +224,7 @@ public class DumpCat {
vcSerializable.readFromStream(input);
if (batchNum != targetBatchNum) {
- VectorContainer vectorContainer = (VectorContainer) vcSerializable.get();
+ final VectorContainer vectorContainer = (VectorContainer) vcSerializable.get();
vectorContainer.zeroVectors();
}
}
@@ -238,22 +237,21 @@ public class DumpCat {
if (vcSerializable != null) {
showSingleBatch(vcSerializable, showHeader);
- VectorContainer vectorContainer = (VectorContainer) vcSerializable.get();
+ final VectorContainer vectorContainer = (VectorContainer) vcSerializable.get();
vectorContainer.zeroVectors();
}
}
-
private void showSingleBatch (VectorAccessibleSerializable vcSerializable, boolean showHeader) {
- VectorContainer vectorContainer = (VectorContainer)vcSerializable.get();
+ final VectorContainer vectorContainer = (VectorContainer) vcSerializable.get();
/* show the header of the batch */
if (showHeader) {
System.out.println(getBatchMetaInfo(vcSerializable).toString());
System.out.println("Schema Information");
- for (VectorWrapper w : vectorContainer) {
- MaterializedField field = w.getValueVector().getField();
+ for (final VectorWrapper w : vectorContainer) {
+ final MaterializedField field = w.getValueVector().getField();
System.out.println (String.format("name : %s, minor_type : %s, data_mode : %s",
field.toExpr(),
field.getType().getMinorType().toString(),
@@ -266,12 +264,11 @@ public class DumpCat {
VectorUtil.showVectorAccessibleContent(vectorContainer);
}
-
/* Get batch meta info : rows, selectedRows, dataSize */
private BatchMetaInfo getBatchMetaInfo(VectorAccessibleSerializable vcSerializable) {
- VectorAccessible vectorContainer = vcSerializable.get();
+ final VectorAccessible vectorContainer = vcSerializable.get();
- int rows =0;
+ int rows = 0;
int selectedRows = 0;
int totalDataSize = 0;
@@ -282,11 +279,10 @@ public class DumpCat {
selectedRows = vcSerializable.getSv2().getCount();
}
- for (VectorWrapper w : vectorContainer) {
+ for (final VectorWrapper w : vectorContainer) {
totalDataSize += w.getValueVector().getBufferSize();
}
- return new BatchMetaInfo(rows, selectedRows, totalDataSize);
+ return new BatchMetaInfo(rows, selectedRows, totalDataSize);
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java
index 8cb4a0b..212e8b0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/PrintingResultsListener.java
@@ -28,7 +28,7 @@ import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.client.QuerySubmitter.Format;
import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.proto.UserBitShared.QueryData;
import org.apache.drill.exec.proto.UserBitShared.QueryId;
import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
@@ -54,7 +54,7 @@ public class PrintingResultsListener implements UserResultsListener {
Stopwatch w = new Stopwatch();
public PrintingResultsListener(DrillConfig config, Format format, int columnWidth) {
- this.allocator = new TopLevelAllocator(config);
+ this.allocator = RootAllocatorFactory.newRoot(config);
loader = new RecordBatchLoader(allocator);
this.format = format;
this.columnWidth = columnWidth;
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/main/java/org/apache/drill/exec/memory/RootAllocatorFactory.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/memory/RootAllocatorFactory.java b/exec/java-exec/src/main/java/org/apache/drill/exec/memory/RootAllocatorFactory.java
new file mode 100644
index 0000000..a413e4a
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/memory/RootAllocatorFactory.java
@@ -0,0 +1,67 @@
+/**
+ * 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.memory;
+
+import org.apache.drill.common.config.DrillConfig;
+
+import com.google.common.annotations.VisibleForTesting;
+
+public class RootAllocatorFactory {
+ /**
+ * Constructor to prevent instantiation of this static utility class.
+ */
+ private RootAllocatorFactory() {}
+
+ /**
+ * Factory method.
+ *
+ * @param drillConfig the DrillConfig
+ * @return a new root allocator
+ */
+ public static BufferAllocator newRoot(final DrillConfig drillConfig) {
+/* TODO(DRILL-1942)
+ if (BaseAllocator.DEBUG) {
+ return new TopLevelAllocator(drillConfig);
+ }
+*/
+
+ return new TopLevelAllocator(drillConfig);
+ }
+
+ /**
+ * Factory method for testing, where a DrillConfig may not be available.
+ *
+ * @param allocationPolicy the allocation policy
+ * @param initAllocation initial allocation from parent
+ * @param maxReservation maximum amount of memory this can hand out
+ * @param flags one of BufferAllocator's F_* flags
+ * @return a new root allocator
+ */
+/* TODO(DRILL-1942)
+ @VisibleForTesting
+ public static BufferAllocator newRoot(
+ final AllocationPolicy allocationPolicy,
+ final long initAllocation, final long maxReservation, final int flags) {
+ if (BaseAllocator.DEBUG) {
+ return new RootAllocatorDebug(allocationPolicy, initAllocation, maxReservation, flags);
+ }
+
+ return new RootAllocator(allocationPolicy, initAllocation, maxReservation, flags);
+ }
+*/
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/main/java/org/apache/drill/exec/memory/TopLevelAllocator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/memory/TopLevelAllocator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/memory/TopLevelAllocator.java
index 05849ea..6e23402 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/memory/TopLevelAllocator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/memory/TopLevelAllocator.java
@@ -51,15 +51,17 @@ public class TopLevelAllocator implements BufferAllocator {
private final DrillBuf empty;
private final DrillConfig config;
+ /* TODO(cwestin) remove
@Deprecated
- public TopLevelAllocator() {
+ TopLevelAllocator() {
this(DrillConfig.getMaxDirectMemory());
}
@Deprecated
- public TopLevelAllocator(long maximumAllocation) {
+ TopLevelAllocator(long maximumAllocation) {
this(null, maximumAllocation, true);
}
+ */
private TopLevelAllocator(DrillConfig config, long maximumAllocation, boolean errorOnLeak){
MAXIMUM_DIRECT_MEMORY = maximumAllocation;
@@ -70,7 +72,7 @@ public class TopLevelAllocator implements BufferAllocator {
this.childrenMap = ENABLE_ACCOUNTING ? new IdentityHashMap<ChildAllocator, StackTraceElement[]>() : null;
}
- public TopLevelAllocator(DrillConfig config) {
+ TopLevelAllocator(DrillConfig config) {
this(config, Math.min(DrillConfig.getMaxDirectMemory(), config.getLong(ExecConstants.TOP_LEVEL_MAX_ALLOC)),
config.getBoolean(ExecConstants.ERROR_ON_MEMORY_LEAK)
);
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java
index d0a998e..120c0d0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java
@@ -24,15 +24,15 @@ import java.io.Closeable;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.metrics.DrillMetrics;
import org.apache.drill.exec.rpc.TransportCheck;
import com.codahale.metrics.MetricRegistry;
// TODO: Doc. What kind of context? (For what aspects, RPC? What kind of data?)
-public class BootStrapContext implements Closeable{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BootStrapContext.class);
+public class BootStrapContext implements Closeable {
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BootStrapContext.class);
private final DrillConfig config;
private final EventLoopGroup loop;
@@ -41,12 +41,11 @@ public class BootStrapContext implements Closeable{
private final BufferAllocator allocator;
public BootStrapContext(DrillConfig config) {
- super();
this.config = config;
this.loop = TransportCheck.createEventLoopGroup(config.getInt(ExecConstants.BIT_SERVER_RPC_THREADS), "BitServer-");
this.loop2 = TransportCheck.createEventLoopGroup(config.getInt(ExecConstants.BIT_SERVER_RPC_THREADS), "BitClient-");
this.metrics = DrillMetrics.getInstance();
- this.allocator = new TopLevelAllocator(config);
+ this.allocator = RootAllocatorFactory.newRoot(config);
}
public DrillConfig getConfig() {
@@ -69,10 +68,10 @@ public class BootStrapContext implements Closeable{
return allocator;
}
+ @Override
public void close() {
DrillMetrics.resetMetrics();
loop.shutdownGracefully();
allocator.close();
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/BaseTestQuery.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/BaseTestQuery.java b/exec/java-exec/src/test/java/org/apache/drill/BaseTestQuery.java
index d428920..eaf8765 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/BaseTestQuery.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/BaseTestQuery.java
@@ -27,6 +27,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.io.Files;
+
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.ExecConstants;
@@ -34,7 +35,7 @@ import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.client.DrillClient;
import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.proto.UserBitShared;
import org.apache.drill.exec.proto.UserBitShared.QueryId;
import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
@@ -166,7 +167,7 @@ public class BaseTestQuery extends ExecTest {
}
private static void openClient() throws Exception {
- allocator = new TopLevelAllocator(config);
+ allocator = RootAllocatorFactory.newRoot(config);
if (config.hasPath(ENABLE_FULL_CACHE) && config.getBoolean(ENABLE_FULL_CACHE)) {
serviceSet = RemoteServiceSet.getServiceSetWithFullCache(config, allocator);
} else {
@@ -228,13 +229,13 @@ public class BaseTestQuery extends ExecTest {
}
@AfterClass
- public static void closeClient() throws IOException, InterruptedException {
+ public static void closeClient() throws IOException {
if (client != null) {
client.close();
}
if (bits != null) {
- for(Drillbit bit : bits) {
+ for(final Drillbit bit : bits) {
if (bit != null) {
bit.close();
}
@@ -256,7 +257,7 @@ public class BaseTestQuery extends ExecTest {
}
protected static void runSQL(String sql) throws Exception {
- SilentListener listener = new SilentListener();
+ final SilentListener listener = new SilentListener();
testWithListener(QueryType.SQL, sql, listener);
listener.waitForCompletion();
}
@@ -294,8 +295,8 @@ public class BaseTestQuery extends ExecTest {
query = String.format(query, args);
logger.debug("Running query:\n--------------\n" + query);
for (int i = 0; i < interation; i++) {
- List<QueryDataBatch> results = client.runQuery(QueryType.SQL, query);
- for (QueryDataBatch queryDataBatch : results) {
+ final List<QueryDataBatch> results = client.runQuery(QueryType.SQL, query);
+ for (final QueryDataBatch queryDataBatch : results) {
queryDataBatch.release();
}
}
@@ -366,7 +367,7 @@ public class BaseTestQuery extends ExecTest {
}
public static String getFile(String resource) throws IOException{
- URL url = Resources.getResource(resource);
+ final URL url = Resources.getResource(resource);
if (url == null) {
throw new IOException(String.format("Unable to find path %s.", resource));
}
@@ -376,13 +377,13 @@ public class BaseTestQuery extends ExecTest {
/**
* Copy the resource (ex. file on classpath) to a physical file on FileSystem.
* @param resource
- * @return
+ * @return the file path
* @throws IOException
*/
public static String getPhysicalFileFromResource(final String resource) throws IOException {
final File file = File.createTempFile("tempfile", ".txt");
file.deleteOnExit();
- PrintWriter printWriter = new PrintWriter(file);
+ final PrintWriter printWriter = new PrintWriter(file);
printWriter.write(BaseTestQuery.getFile(resource));
printWriter.close();
@@ -395,7 +396,7 @@ public class BaseTestQuery extends ExecTest {
* @return Full path including temp parent directory and given directory name.
*/
public static String getTempDir(final String dirName) {
- File dir = Files.createTempDir();
+ final File dir = Files.createTempDir();
dir.deleteOnExit();
return dir.getAbsolutePath() + File.separator + dirName;
@@ -412,8 +413,8 @@ public class BaseTestQuery extends ExecTest {
private static class SilentListener implements UserResultsListener {
private volatile UserException exception;
- private AtomicInteger count = new AtomicInteger();
- private CountDownLatch latch = new CountDownLatch(1);
+ private final AtomicInteger count = new AtomicInteger();
+ private final CountDownLatch latch = new CountDownLatch(1);
@Override
public void submissionFailed(UserException ex) {
@@ -430,7 +431,7 @@ public class BaseTestQuery extends ExecTest {
@Override
public void dataArrived(QueryDataBatch result, ConnectionThrottle throttle) {
- int rows = result.getHeader().getRowCount();
+ final int rows = result.getHeader().getRowCount();
if (result.getData() != null) {
count.addAndGet(rows);
}
@@ -459,8 +460,8 @@ public class BaseTestQuery extends ExecTest {
protected int printResult(List<QueryDataBatch> results) throws SchemaChangeException {
int rowCount = 0;
- RecordBatchLoader loader = new RecordBatchLoader(getAllocator());
- for(QueryDataBatch result : results) {
+ final RecordBatchLoader loader = new RecordBatchLoader(getAllocator());
+ for(final QueryDataBatch result : results) {
rowCount += result.getHeader().getRowCount();
loader.load(result.getHeader().getDef(), result.getData());
// TODO: Clean: DRILL-2933: That load(...) no longer throws
@@ -478,10 +479,10 @@ public class BaseTestQuery extends ExecTest {
protected static String getResultString(List<QueryDataBatch> results, String delimiter)
throws SchemaChangeException {
- StringBuilder formattedResults = new StringBuilder();
+ final StringBuilder formattedResults = new StringBuilder();
boolean includeHeader = true;
- RecordBatchLoader loader = new RecordBatchLoader(getAllocator());
- for(QueryDataBatch result : results) {
+ final RecordBatchLoader loader = new RecordBatchLoader(getAllocator());
+ for(final QueryDataBatch result : results) {
loader.load(result.getHeader().getDef(), result.getData());
if (loader.getRecordCount() <= 0) {
continue;
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/PlanningBase.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/PlanningBase.java b/exec/java-exec/src/test/java/org/apache/drill/PlanningBase.java
index da033f7..05b7238 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/PlanningBase.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/PlanningBase.java
@@ -22,14 +22,15 @@ import java.net.URL;
import mockit.Mocked;
import mockit.NonStrictExpectations;
+
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.jdbc.SimpleCalciteSchema;
-
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.util.TestTools;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.QueryContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.planner.physical.PlannerSettings;
@@ -54,7 +55,7 @@ import com.google.common.collect.ImmutableList;
import com.google.common.io.Resources;
public class PlanningBase extends ExecTest{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PlanningBase.class);
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PlanningBase.class);
@Rule public final TestRule TIMEOUT = TestTools.getTimeoutRule(10000);
@@ -63,16 +64,14 @@ public class PlanningBase extends ExecTest{
@Mocked QueryContext context;
- TopLevelAllocator allocator = new TopLevelAllocator();
+ BufferAllocator allocator = RootAllocatorFactory.newRoot(config);
protected void testSqlPlanFromFile(String file) throws Exception {
testSqlPlan(getFile(file));
}
protected void testSqlPlan(String sqlCommands) throws Exception {
- String[] sqlStrings = sqlCommands.split(";");
-
-
+ final String[] sqlStrings = sqlCommands.split(";");
final LocalPStoreProvider provider = new LocalPStoreProvider(config);
provider.start();
@@ -134,18 +133,17 @@ public class PlanningBase extends ExecTest{
}
};
- for (String sql : sqlStrings) {
+ for (final String sql : sqlStrings) {
if (sql.trim().isEmpty()) {
continue;
}
- DrillSqlWorker worker = new DrillSqlWorker(context);
- PhysicalPlan p = worker.getPlan(sql);
+ final DrillSqlWorker worker = new DrillSqlWorker(context);
+ final PhysicalPlan p = worker.getPlan(sql);
}
-
}
- protected String getFile(String resource) throws IOException{
- URL url = Resources.getResource(resource);
+ protected static String getFile(String resource) throws IOException {
+ final URL url = Resources.getResource(resource);
if (url == null) {
throw new IOException(String.format("Unable to find path %s.", resource));
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/client/DumpCatTest.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/client/DumpCatTest.java b/exec/java-exec/src/test/java/org/apache/drill/exec/client/DumpCatTest.java
index 7c58b19..6b6bd64 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/client/DumpCatTest.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/client/DumpCatTest.java
@@ -30,7 +30,7 @@ import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -58,9 +58,9 @@ import com.google.common.io.Files;
* which will produce a dump file. The dump file will be input into DumpCat to test query mode and batch mode.
*/
-public class DumpCatTest extends ExecTest{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DumpCatTest.class);
- DrillConfig c = DrillConfig.create();
+public class DumpCatTest extends ExecTest {
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DumpCatTest.class);
+ private final DrillConfig c = DrillConfig.create();
@Test
public void testDumpCat(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable
@@ -68,22 +68,22 @@ public class DumpCatTest extends ExecTest{
new NonStrictExpectations(){{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/trace/simple_trace.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/trace/simple_trace.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- while(exec.next()){
+ while(exec.next()) {
}
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
@@ -95,37 +95,34 @@ public class DumpCatTest extends ExecTest{
/* Form the file name to which the trace output will dump the record batches */
String qid = QueryIdHelper.getQueryId(handle.getQueryId());
- int majorFragmentId = handle.getMajorFragmentId();
- int minorFragmentId = handle.getMinorFragmentId();
+ final int majorFragmentId = handle.getMajorFragmentId();
+ final int minorFragmentId = handle.getMinorFragmentId();
- String logLocation = c.getString(ExecConstants.TRACE_DUMP_DIRECTORY);
+ final String logLocation = c.getString(ExecConstants.TRACE_DUMP_DIRECTORY);
System.out.println("Found log location: " + logLocation);
- String filename = String.format("%s//%s_%d_%d_mock-scan", logLocation, qid, majorFragmentId, minorFragmentId);
+ final String filename = String.format("%s//%s_%d_%d_mock-scan", logLocation, qid, majorFragmentId, minorFragmentId);
System.out.println("File Name: " + filename);
- Configuration conf = new Configuration();
+ final Configuration conf = new Configuration();
conf.set(FileSystem.FS_DEFAULT_NAME_KEY, c.getString(ExecConstants.TRACE_DUMP_FILESYSTEM));
- FileSystem fs = FileSystem.get(conf);
- Path path = new Path(filename);
+ final FileSystem fs = FileSystem.get(conf);
+ final Path path = new Path(filename);
assertTrue("Trace file does not exist", fs.exists(path));
- DumpCat dumpCat = new DumpCat();
+ final DumpCat dumpCat = new DumpCat();
//Test Query mode
- FileInputStream input = new FileInputStream(filename);
-
- dumpCat.doQuery(input);
- input.close();
+ try (final FileInputStream input = new FileInputStream(filename)) {
+ dumpCat.doQuery(input);
+ }
//Test Batch mode
- input = new FileInputStream(filename);
- dumpCat.doBatch(input,0,true);
-
- input.close();
+ try(final FileInputStream input = new FileInputStream(filename)) {
+ dumpCat.doBatch(input, 0, true);
+ }
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/expr/ExpressionTest.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/expr/ExpressionTest.java b/exec/java-exec/src/test/java/org/apache/drill/exec/expr/ExpressionTest.java
index 5e50189..cf6fb69 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/expr/ExpressionTest.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/expr/ExpressionTest.java
@@ -36,14 +36,12 @@ import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.expression.parser.ExprLexer;
import org.apache.drill.common.expression.parser.ExprParser;
import org.apache.drill.common.expression.parser.ExprParser.parse_return;
-import org.apache.drill.common.types.MajorType;
import org.apache.drill.common.types.TypeProtos;
import org.apache.drill.common.types.TypeProtos.MinorType;
import org.apache.drill.common.types.Types;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.physical.impl.project.Projector;
import org.apache.drill.exec.record.MaterializedField;
import org.apache.drill.exec.record.RecordBatch;
@@ -54,10 +52,10 @@ import org.apache.drill.exec.vector.ValueVector;
import org.junit.Test;
public class ExpressionTest extends ExecTest {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExpressionTest.class);
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExpressionTest.class);
- DrillConfig c = DrillConfig.create();
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ private final DrillConfig c = DrillConfig.create();
+ private final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
@Test
public void testBasicExpression(@Injectable RecordBatch batch) throws Exception {
@@ -87,7 +85,7 @@ public class ExpressionTest extends ExecTest {
batch.getValueAccessorById(IntVector.class, tfid.getFieldIds());
result = wrapper;
wrapper.getValueVector();
- result = new IntVector(MaterializedField.create("result", type), new TopLevelAllocator(0));
+ result = new IntVector(MaterializedField.create("result", type), RootAllocatorFactory.newRoot(c));
}
};
@@ -123,25 +121,24 @@ public class ExpressionTest extends ExecTest {
// HELPER METHODS //
private LogicalExpression parseExpr(String expr) throws RecognitionException {
- ExprLexer lexer = new ExprLexer(new ANTLRStringStream(expr));
- CommonTokenStream tokens = new CommonTokenStream(lexer);
- ExprParser parser = new ExprParser(tokens);
+ final ExprLexer lexer = new ExprLexer(new ANTLRStringStream(expr));
+ final CommonTokenStream tokens = new CommonTokenStream(lexer);
+ final ExprParser parser = new ExprParser(tokens);
parse_return ret = parser.parse();
return ret.e;
}
private String getExpressionCode(String expression, RecordBatch batch) throws Exception {
- LogicalExpression expr = parseExpr(expression);
- ErrorCollector error = new ErrorCollectorImpl();
- LogicalExpression materializedExpr = ExpressionTreeMaterializer.materialize(expr, batch, error, registry);
+ final LogicalExpression expr = parseExpr(expression);
+ final ErrorCollector error = new ErrorCollectorImpl();
+ final LogicalExpression materializedExpr = ExpressionTreeMaterializer.materialize(expr, batch, error, registry);
if (error.getErrorCount() != 0) {
logger.error("Failure while materializing expression [{}]. Errors: {}", expression, error);
assertEquals(0, error.getErrorCount());
}
- ClassGenerator<Projector> cg = CodeGenerator.get(Projector.TEMPLATE_DEFINITION, new FunctionImplementationRegistry(DrillConfig.create())).getRoot();
+ final ClassGenerator<Projector> cg = CodeGenerator.get(Projector.TEMPLATE_DEFINITION, new FunctionImplementationRegistry(DrillConfig.create())).getRoot();
cg.addExpr(new ValueVectorWriteExpression(new TypedFieldId(materializedExpr.getMajorType(), -1), materializedExpr));
return cg.getCodeGenerator().generateAndGet();
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestByteComparisonFunctions.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestByteComparisonFunctions.java b/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestByteComparisonFunctions.java
index 1e9a47c..d8ce7c2 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestByteComparisonFunctions.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestByteComparisonFunctions.java
@@ -23,25 +23,26 @@ import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers;
import org.apache.drill.exec.expr.holders.VarCharHolder;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.vector.ValueHolderHelper;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
-public class TestByteComparisonFunctions extends ExecTest{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestByteComparisonFunctions.class);
+public class TestByteComparisonFunctions extends ExecTest {
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestByteComparisonFunctions.class);
- static TopLevelAllocator allocator;
- static VarCharHolder hello;
- static VarCharHolder goodbye;
- static VarCharHolder helloLong;
- static VarCharHolder goodbyeLong;
+ private static BufferAllocator allocator;
+ private static VarCharHolder hello;
+ private static VarCharHolder goodbye;
+ private static VarCharHolder helloLong;
+ private static VarCharHolder goodbyeLong;
@BeforeClass
- public static void setup(){
+ public static void setup() {
DrillConfig c= DrillConfig.create();
- allocator = new TopLevelAllocator(c);
+ allocator = RootAllocatorFactory.newRoot(c);
hello = ValueHolderHelper.getVarCharHolder(allocator, "hello");
goodbye = ValueHolderHelper.getVarCharHolder(allocator, "goodbye");
helloLong = ValueHolderHelper.getVarCharHolder(allocator, "hellomyfriend");
@@ -49,7 +50,7 @@ public class TestByteComparisonFunctions extends ExecTest{
}
@AfterClass
- public static void teardown(){
+ public static void teardown() {
hello.buffer.release();
helloLong.buffer.release();
goodbye.buffer.release();
@@ -58,72 +59,72 @@ public class TestByteComparisonFunctions extends ExecTest{
}
@Test
- public void testAfter(){
- VarCharHolder left = hello;
- VarCharHolder right = goodbye;
+ public void testAfter() {
+ final VarCharHolder left = hello;
+ final VarCharHolder right = goodbye;
assertTrue(ByteFunctionHelpers.compare(left.buffer, left.start, left.end, right.buffer, right.start, right.end) == 1);
}
@Test
- public void testBefore(){
- VarCharHolder left = goodbye;
- VarCharHolder right = hello;
+ public void testBefore() {
+ final VarCharHolder left = goodbye;
+ final VarCharHolder right = hello;
assertTrue(ByteFunctionHelpers.compare(left.buffer, left.start, left.end, right.buffer, right.start, right.end) == -1);
}
@Test
- public void testEqualCompare(){
- VarCharHolder left = hello;
- VarCharHolder right = hello;
+ public void testEqualCompare() {
+ final VarCharHolder left = hello;
+ final VarCharHolder right = hello;
assertTrue(ByteFunctionHelpers.compare(left.buffer, left.start, left.end, right.buffer, right.start, right.end) == 0);
}
@Test
- public void testEqual(){
- VarCharHolder left = hello;
- VarCharHolder right = hello;
+ public void testEqual() {
+ final VarCharHolder left = hello;
+ final VarCharHolder right = hello;
assertTrue(ByteFunctionHelpers.equal(left.buffer, left.start, left.end, right.buffer, right.start, right.end) == 1);
}
@Test
- public void testNotEqual(){
- VarCharHolder left = hello;
- VarCharHolder right = goodbye;
+ public void testNotEqual() {
+ final VarCharHolder left = hello;
+ final VarCharHolder right = goodbye;
assertTrue(ByteFunctionHelpers.equal(left.buffer, left.start, left.end, right.buffer, right.start, right.end) == 0);
}
@Test
- public void testAfterLong(){
- VarCharHolder left = helloLong;
- VarCharHolder right = goodbyeLong;
+ public void testAfterLong() {
+ final VarCharHolder left = helloLong;
+ final VarCharHolder right = goodbyeLong;
assertTrue(ByteFunctionHelpers.compare(left.buffer, left.start, left.end, right.buffer, right.start, right.end) == 1);
}
@Test
- public void testBeforeLong(){
- VarCharHolder left = goodbyeLong;
- VarCharHolder right = helloLong;
+ public void testBeforeLong() {
+ final VarCharHolder left = goodbyeLong;
+ final VarCharHolder right = helloLong;
assertTrue(ByteFunctionHelpers.compare(left.buffer, left.start, left.end, right.buffer, right.start, right.end) == -1);
}
@Test
- public void testEqualCompareLong(){
- VarCharHolder left = helloLong;
- VarCharHolder right = helloLong;
+ public void testEqualCompareLong() {
+ final VarCharHolder left = helloLong;
+ final VarCharHolder right = helloLong;
assertTrue(ByteFunctionHelpers.compare(left.buffer, left.start, left.end, right.buffer, right.start, right.end) == 0);
}
@Test
- public void testEqualLong(){
- VarCharHolder left = helloLong;
- VarCharHolder right = helloLong;
+ public void testEqualLong() {
+ final VarCharHolder left = helloLong;
+ final VarCharHolder right = helloLong;
assertTrue(ByteFunctionHelpers.equal(left.buffer, left.start, left.end, right.buffer, right.start, right.end) == 1);
}
@Test
- public void testNotEqualLong(){
- VarCharHolder left = helloLong;
- VarCharHolder right = goodbyeLong;
+ public void testNotEqualLong() {
+ final VarCharHolder left = helloLong;
+ final VarCharHolder right = goodbyeLong;
assertTrue(ByteFunctionHelpers.equal(left.buffer, left.start, left.end, right.buffer, right.start, right.end) == 0);
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestMathFunctions.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestMathFunctions.java b/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestMathFunctions.java
index 4f06a9d..99ee904 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestMathFunctions.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestMathFunctions.java
@@ -30,7 +30,7 @@ import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -52,44 +52,41 @@ import com.google.common.io.Files;
public class TestMathFunctions extends ExecTest {
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestMathFunctions.class);
+ private final DrillConfig c = DrillConfig.create();
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestMathFunctions.class);
- DrillConfig c = DrillConfig.create();
+ @Test
+ public void testBasicMathFunctions(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable
+ {
+ new NonStrictExpectations() {{
+ bitContext.getMetrics(); result = new MetricRegistry();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
+ bitContext.getConfig(); result = c;
+ bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
+ bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
+ }};
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/simple_math_functions.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, BitControl.PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- @Test
- public void testBasicMathFunctions(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable
- {
-
- new NonStrictExpectations(){{
- bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
- bitContext.getConfig(); result = c;
- bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
- bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
- }};
-
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/simple_math_functions.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, BitControl.PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
-
- while(exec.next()) {
- IntVector intMulVector = exec.getValueVectorById(new SchemaPath("INTMUL", ExpressionPosition.UNKNOWN), IntVector.class);
- Float8Vector floatMulVector = exec.getValueVectorById(new SchemaPath("FLOATMUL", ExpressionPosition.UNKNOWN), Float8Vector.class);
- IntVector intAddVector = exec.getValueVectorById(new SchemaPath("INTADD", ExpressionPosition.UNKNOWN), IntVector.class);
- Float8Vector floatAddVector = exec.getValueVectorById(new SchemaPath("FLOATADD", ExpressionPosition.UNKNOWN), Float8Vector.class);
- assertEquals(exec.getRecordCount(), 1);
- assertEquals(intMulVector.getAccessor().get(0), 2);
- assertEquals(floatMulVector.getAccessor().get(0), (1.1 * 2.2), 0);
- assertEquals(intAddVector.getAccessor().get(0), 3);
- assertEquals(floatAddVector.getAccessor().get(0), (1.1 + 2.2), 0);
- }
+ while(exec.next()) {
+ final IntVector intMulVector = exec.getValueVectorById(new SchemaPath("INTMUL", ExpressionPosition.UNKNOWN), IntVector.class);
+ final Float8Vector floatMulVector = exec.getValueVectorById(new SchemaPath("FLOATMUL", ExpressionPosition.UNKNOWN), Float8Vector.class);
+ final IntVector intAddVector = exec.getValueVectorById(new SchemaPath("INTADD", ExpressionPosition.UNKNOWN), IntVector.class);
+ final Float8Vector floatAddVector = exec.getValueVectorById(new SchemaPath("FLOATADD", ExpressionPosition.UNKNOWN), Float8Vector.class);
+ assertEquals(exec.getRecordCount(), 1);
+ assertEquals(intMulVector.getAccessor().get(0), 2);
+ assertEquals(floatMulVector.getAccessor().get(0), (1.1 * 2.2), 0);
+ assertEquals(intAddVector.getAccessor().get(0), 3);
+ assertEquals(floatAddVector.getAccessor().get(0), (1.1 + 2.2), 0);
+ }
- if(context.getFailureCause() != null){
- throw context.getFailureCause();
- }
- assertTrue(!context.isFailed());
+ if(context.getFailureCause() != null) {
+ throw context.getFailureCause();
}
+ assertTrue(!context.isFailed());
+ }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestNewMathFunctions.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestNewMathFunctions.java b/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestNewMathFunctions.java
index 880184e..3447dae 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestNewMathFunctions.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestNewMathFunctions.java
@@ -29,7 +29,7 @@ import mockit.NonStrictExpectations;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -50,23 +50,21 @@ import com.google.common.base.Charsets;
import com.google.common.io.Resources;
public class TestNewMathFunctions {
-
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestNewMathFunctions.class);
-
- DrillConfig c = DrillConfig.create();
- PhysicalPlanReader reader;
- FunctionImplementationRegistry registry;
- FragmentContext context;
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestNewMathFunctions.class);
+ private final DrillConfig c = DrillConfig.create();
+ private PhysicalPlanReader reader;
+ private FunctionImplementationRegistry registry;
+ private FragmentContext context;
public Object[] getRunResult(SimpleRootExec exec) {
int size = 0;
- for (ValueVector v : exec) {
+ for (final ValueVector v : exec) {
size++;
}
- Object[] res = new Object [size];
+ final Object[] res = new Object[size];
int i = 0;
- for (ValueVector v : exec) {
+ for (final ValueVector v : exec) {
if (v instanceof VarCharVector) {
res[i++] = new String( ((VarCharVector) v).getAccessor().get(0));
} else {
@@ -78,16 +76,15 @@ public class TestNewMathFunctions {
public void runTest(@Injectable final DrillbitContext bitContext,
@Injectable UserServer.UserClientConnection connection, Object[] expectedResults, String planPath) throws Throwable {
-
new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- String planString = Resources.toString(Resources.getResource(planPath), Charsets.UTF_8);
+ final String planString = Resources.toString(Resources.getResource(planPath), Charsets.UTF_8);
if (reader == null) {
reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
}
@@ -97,12 +94,12 @@ public class TestNewMathFunctions {
if (context == null) {
context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry); //new FragmentContext(bitContext, ExecProtos.FragmentHandle.getDefaultInstance(), connection, registry);
}
- PhysicalPlan plan = reader.readPhysicalPlan(planString);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlan plan = reader.readPhysicalPlan(planString);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
exec.next(); // skip schema batch
while (exec.next()) {
- Object [] res = getRunResult(exec);
+ final Object [] res = getRunResult(exec);
assertEquals("return count does not match", expectedResults.length, res.length);
for (int i = 0; i<res.length; i++) {
@@ -119,17 +116,16 @@ public class TestNewMathFunctions {
@Test
public void testTrigoMathFunc(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {Math.sin(45), Math.cos(45), Math.tan(45),Math.asin(45), Math.acos(45), Math.atan(45),Math.sinh(45), Math.cosh(45), Math.tanh(45)};
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {Math.sin(45), Math.cos(45), Math.tan(45),Math.asin(45), Math.acos(45), Math.atan(45),Math.sinh(45), Math.cosh(45), Math.tanh(45)};
runTest(bitContext, connection, expected, "functions/testTrigoMathFunctions.json");
}
@Test
public void testExtendedMathFunc(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- BigDecimal d = new BigDecimal("100111111111111111111111111111111111.00000000000000000000000000000000000000000000000000001");
-
- Object [] expected = new Object[] {Math.cbrt(1000), Math.log(10), (Math.log(64.0)/Math.log(2.0)), Math.exp(10), Math.toDegrees(0.5), Math.toRadians(45.0), Math.PI, Math.cbrt(d.doubleValue()), Math.log(d.doubleValue()), (Math.log(d.doubleValue())/Math.log(2)), Math.exp(d.doubleValue()), Math.toDegrees(d.doubleValue()), Math.toRadians(d.doubleValue())};
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final BigDecimal d = new BigDecimal("100111111111111111111111111111111111.00000000000000000000000000000000000000000000000000001");
+ final Object [] expected = new Object[] {Math.cbrt(1000), Math.log(10), (Math.log(64.0)/Math.log(2.0)), Math.exp(10), Math.toDegrees(0.5), Math.toRadians(45.0), Math.PI, Math.cbrt(d.doubleValue()), Math.log(d.doubleValue()), (Math.log(d.doubleValue())/Math.log(2)), Math.exp(d.doubleValue()), Math.toDegrees(d.doubleValue()), Math.toRadians(d.doubleValue())};
runTest(bitContext, connection, expected, "functions/testExtendedMathFunctions.json");
}
@@ -137,15 +133,14 @@ public class TestNewMathFunctions {
@Test
public void testTruncDivMod(@Injectable final DrillbitContext bitContext,
@Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {101.0, 0, 101, 1010.0, 101, 481.0, 0.001099999999931267};
+ final Object [] expected = new Object[] {101.0, 0, 101, 1010.0, 101, 481.0, 0.001099999999931267};
runTest(bitContext, connection, expected, "functions/testDivModTruncFunctions.json");
}
@Test
public void testIsNumeric(@Injectable final DrillbitContext bitContext,
@Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {1, 1, 1, 0};
+ final Object [] expected = new Object[] {1, 1, 1, 0};
runTest(bitContext, connection, expected, "functions/testIsNumericFunction.json");
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestRepeatedFunction.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestRepeatedFunction.java b/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestRepeatedFunction.java
index 73c7508..68c3b9c 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestRepeatedFunction.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestRepeatedFunction.java
@@ -29,7 +29,7 @@ import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -50,42 +50,38 @@ import com.google.common.base.Charsets;
import com.google.common.io.Files;
public class TestRepeatedFunction extends ExecTest{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestRepeatedFunction.class);
- DrillConfig c = DrillConfig.create();
-
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestRepeatedFunction.class);
+ private final DrillConfig c = DrillConfig.create();
@Test
- public void testRepeated(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable{
+ public void testRepeated(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable {
// System.out.println(System.getProperty("java.class.path"));
-
-
- new NonStrictExpectations(){{
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
-
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/physical_repeated_1.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/physical_repeated_1.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
boolean oneIsOne = false;
int size = 0;
- int[] sizes = {1,2,0,6};
+ final int[] sizes = {1, 2, 0, 6};
- while(exec.next()){
- IntVector c1 = exec.getValueVectorById(new SchemaPath("cnt", ExpressionPosition.UNKNOWN), IntVector.class);
- BitVector c2 = exec.getValueVectorById(new SchemaPath("has_min", ExpressionPosition.UNKNOWN), BitVector.class);
+ while(exec.next()) {
+ final IntVector c1 = exec.getValueVectorById(new SchemaPath("cnt", ExpressionPosition.UNKNOWN), IntVector.class);
+ final BitVector c2 = exec.getValueVectorById(new SchemaPath("has_min", ExpressionPosition.UNKNOWN), BitVector.class);
- for(int i =0; i < exec.getRecordCount(); i++){
- int curSize = sizes[size % sizes.length];
+ for(int i = 0; i < exec.getRecordCount(); i++) {
+ final int curSize = sizes[size % sizes.length];
assertEquals(curSize, c1.getAccessor().get(i));
- switch(curSize){
+ switch(curSize) {
case 1:
assertEquals(oneIsOne, 1 == c2.getAccessor().get(i));
oneIsOne = !oneIsOne;
@@ -104,11 +100,9 @@ public class TestRepeatedFunction extends ExecTest{
}
}
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
-
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/memory/TestEndianess.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/memory/TestEndianess.java b/exec/java-exec/src/test/java/org/apache/drill/exec/memory/TestEndianess.java
index 48ddada..f6fe79e 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/memory/TestEndianess.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/memory/TestEndianess.java
@@ -20,22 +20,23 @@ package org.apache.drill.exec.memory;
import static org.junit.Assert.assertEquals;
import io.netty.buffer.ByteBuf;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecTest;
import org.junit.Test;
-
-public class TestEndianess extends ExecTest{
-
+public class TestEndianess extends ExecTest {
@Test
- public void testLittleEndian(){
- TopLevelAllocator a = new TopLevelAllocator(Long.MAX_VALUE);
- ByteBuf b = a.buffer(4);
+ public void testLittleEndian() {
+ final DrillConfig drillConfig = DrillConfig.create();
+ final BufferAllocator a = RootAllocatorFactory.newRoot(drillConfig);
+ final ByteBuf b = a.buffer(4);
b.setInt(0, 35);
- assertEquals((int) b.getByte(0), 35);
- assertEquals((int) b.getByte(1), 0);
- assertEquals((int) b.getByte(2), 0);
- assertEquals((int) b.getByte(3), 0);
+ assertEquals(b.getByte(0), 35);
+ assertEquals(b.getByte(1), 0);
+ assertEquals(b.getByte(2), 0);
+ assertEquals(b.getByte(3), 0);
+ b.release();
+ a.close();
}
-
}
[2/5] drill git commit: DRILL-3598: use a factory to create the root
allocator. - made the constructor for TopLevelAllocator package private to
enforce this
Posted by js...@apache.org.
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestStringFunctions.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestStringFunctions.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestStringFunctions.java
index d72c1e1..32ed220 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestStringFunctions.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestStringFunctions.java
@@ -26,7 +26,7 @@ import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -44,22 +44,22 @@ import com.google.common.base.Charsets;
import com.google.common.io.Resources;
public class TestStringFunctions extends ExecTest {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestStringFunctions.class);
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestStringFunctions.class);
- DrillConfig c = DrillConfig.create();
- PhysicalPlanReader reader;
- FunctionImplementationRegistry registry;
- FragmentContext context;
+ private final DrillConfig c = DrillConfig.create();
+ private PhysicalPlanReader reader;
+ private FunctionImplementationRegistry registry;
+ private FragmentContext context;
public Object[] getRunResult(SimpleRootExec exec) {
int size = 0;
- for (ValueVector v : exec) {
+ for (final ValueVector v : exec) {
size++;
}
- Object[] res = new Object [size];
+ final Object[] res = new Object [size];
int i = 0;
- for (ValueVector v : exec) {
+ for (final ValueVector v : exec) {
if (v instanceof VarCharVector) {
res[i++] = new String( ((VarCharVector) v).getAccessor().get(0), Charsets.UTF_8);
} else {
@@ -73,13 +73,13 @@ public class TestStringFunctions extends ExecTest {
@Injectable UserServer.UserClientConnection connection, Object[] expectedResults, String planPath) throws Throwable {
new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- String planString = Resources.toString(Resources.getResource(planPath), Charsets.UTF_8);
+ final String planString = Resources.toString(Resources.getResource(planPath), Charsets.UTF_8);
if (reader == null) {
reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
}
@@ -89,12 +89,12 @@ public class TestStringFunctions extends ExecTest {
if (context == null) {
context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry); //new FragmentContext(bitContext, ExecProtos.FragmentHandle.getDefaultInstance(), connection, registry);
}
- PhysicalPlan plan = reader.readPhysicalPlan(planString);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlan plan = reader.readPhysicalPlan(planString);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
exec.next(); // skip schema batch
while(exec.next()) {
- Object [] res = getRunResult(exec);
+ final Object [] res = getRunResult(exec);
assertEquals("return count does not match", expectedResults.length, res.length);
for (int i = 0; i<res.length; i++) {
@@ -110,148 +110,130 @@ public class TestStringFunctions extends ExecTest {
@Test
public void testCharLength(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
Object [] expected = new Object[] {new Long(8), new Long(0), new Long(5), new Long(5),
new Long(8), new Long(0), new Long(5), new Long(5),
new Long(8), new Long(0), new Long(5), new Long(5),};
-
runTest(bitContext, connection, expected, "functions/string/testCharLength.json");
}
@Test
public void testLike(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {Boolean.TRUE, Boolean.TRUE, Boolean.TRUE, Boolean.FALSE};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {Boolean.TRUE, Boolean.TRUE, Boolean.TRUE, Boolean.FALSE};
runTest(bitContext, connection, expected, "functions/string/testLike.json");
}
@Test
public void testSimilar(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {Boolean.TRUE, Boolean.FALSE, Boolean.TRUE, Boolean.FALSE};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {Boolean.TRUE, Boolean.FALSE, Boolean.TRUE, Boolean.FALSE};
runTest(bitContext, connection, expected, "functions/string/testSimilar.json");
}
@Test
public void testLtrim(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {"def", "abcdef", "dabc", "", "", ""};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {"def", "abcdef", "dabc", "", "", ""};
runTest(bitContext, connection, expected, "functions/string/testLtrim.json");
}
@Test
public void testTrim(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {"fghI", "", "", "!", " aaa "};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {"fghI", "", "", "!", " aaa "};
runTest(bitContext, connection, expected, "functions/string/testTrim.json");
}
@Test
public void testReplace(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {"aABABcdf", "ABABbABbcdf", "aababcdf", "acdf", "ABCD", "abc"};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {"aABABcdf", "ABABbABbcdf", "aababcdf", "acdf", "ABCD", "abc"};
runTest(bitContext, connection, expected, "functions/string/testReplace.json");
}
@Test
public void testRtrim(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {"abc", "abcdef", "ABd", "", "", ""};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {"abc", "abcdef", "ABd", "", "", ""};
runTest(bitContext, connection, expected, "functions/string/testRtrim.json");
}
@Test
public void testConcat(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {"abcABC", "abc", "ABC", ""};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {"abcABC", "abc", "ABC", ""};
runTest(bitContext, connection, expected, "functions/string/testConcat.json");
}
@Test
public void testLower(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {"abcefgh", "abc", ""};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {"abcefgh", "abc", ""};
runTest(bitContext, connection, expected, "functions/string/testLower.json");
}
@Test
public void testPosition(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {new Long(2), new Long(0), new Long(0), new Long(0),
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {new Long(2), new Long(0), new Long(0), new Long(0),
new Long(2), new Long(0), new Long(0), new Long(0)};
-
runTest(bitContext, connection, expected, "functions/string/testPosition.json");
}
@Test
public void testRight(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {"ef", "abcdef", "abcdef", "cdef", "f", "", ""};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {"ef", "abcdef", "abcdef", "cdef", "f", "", ""};
runTest(bitContext, connection, expected, "functions/string/testRight.json");
}
@Test
public void testSubstr(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {"abc", "bcd", "bcdef", "bcdef", "", "", "", "", "भारत", "वर्ष", "वर्ष", "cdef", "", "", "", "ड्रिल"};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {"abc", "bcd", "bcdef", "bcdef", "", "", "", "", "भारत", "वर्ष", "वर्ष", "cdef", "", "", "", "ड्रिल"};
runTest(bitContext, connection, expected, "functions/string/testSubstr.json");
}
@Test
public void testLeft(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {"ab", "abcdef", "abcdef", "abcd", "a", "", ""};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {"ab", "abcdef", "abcdef", "abcd", "a", "", ""};
runTest(bitContext, connection, expected, "functions/string/testLeft.json");
}
@Test
public void testLpad(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {"", "", "abcdef", "ab", "ab", "abcdef", "AAAAabcdef", "ABABabcdef", "ABCAabcdef", "ABCDabcdef"};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {"", "", "abcdef", "ab", "ab", "abcdef", "AAAAabcdef", "ABABabcdef", "ABCAabcdef", "ABCDabcdef"};
runTest(bitContext, connection, expected, "functions/string/testLpad.json");
}
@Test
public void testRegexpReplace(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {"ThM", "Th", "Thomas"};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {"ThM", "Th", "Thomas"};
runTest(bitContext, connection, expected, "functions/string/testRegexpReplace.json");
}
@Test
public void testRpad(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {"", "", "abcdef", "ab", "ab", "abcdef", "abcdefAAAA", "abcdefABAB", "abcdefABCA", "abcdefABCD"};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {"", "", "abcdef", "ab", "ab", "abcdef", "abcdefAAAA", "abcdefABAB", "abcdefABCA", "abcdefABCD"};
runTest(bitContext, connection, expected, "functions/string/testRpad.json");
}
@Test
public void testUpper(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {"ABCEFGH", "ABC", ""};
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {"ABCEFGH", "ABC", ""};
runTest(bitContext, connection, expected, "functions/string/testUpper.json");
}
@Test
- public void testNewStringFuncs(@Injectable final DrillbitContext bitContext, @Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[] {97, 65, -32, "A", "btrim", "Peace Peace Peace ", "हकुना मताता हकुना मताता ", "katcit", "\u00C3\u00A2pple", "नदम"};
+ public void testNewStringFuncs(@Injectable final DrillbitContext bitContext, @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final Object [] expected = new Object[] {97, 65, -32, "A", "btrim", "Peace Peace Peace ", "हकुना मताता हकुना मताता ", "katcit", "\u00C3\u00A2pple", "नदम"};
runTest(bitContext, connection, expected, "functions/string/testStringFuncs.json");
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestAgg.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestAgg.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestAgg.java
index d2616a8..bf1f064 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestAgg.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestAgg.java
@@ -28,6 +28,7 @@ import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.memory.TopLevelAllocator;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
@@ -50,43 +51,43 @@ import com.google.common.base.Charsets;
import com.google.common.io.Files;
public class TestAgg extends ExecTest {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestAgg.class);
- DrillConfig c = DrillConfig.create();
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestAgg.class);
+ private final DrillConfig c = DrillConfig.create();
- private SimpleRootExec doTest(final DrillbitContext bitContext, UserClientConnection connection, String file) throws Exception{
- new NonStrictExpectations(){{
+ private SimpleRootExec doTest(final DrillbitContext bitContext, UserClientConnection connection, String file) throws Exception {
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile(file), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile(file), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
return exec;
}
@Test
- public void oneKeyAgg(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable{
- SimpleRootExec exec = doTest(bitContext, connection, "/agg/test1.json");
+ public void oneKeyAgg(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable {
+ final SimpleRootExec exec = doTest(bitContext, connection, "/agg/test1.json");
- while(exec.next()){
- BigIntVector cnt = exec.getValueVectorById(SchemaPath.getSimplePath("cnt"), BigIntVector.class);
- IntVector key = exec.getValueVectorById(SchemaPath.getSimplePath("blue"), IntVector.class);
- long[] cntArr = {10001, 9999};
- int[] keyArr = {Integer.MIN_VALUE, Integer.MAX_VALUE};
+ while(exec.next()) {
+ final BigIntVector cnt = exec.getValueVectorById(SchemaPath.getSimplePath("cnt"), BigIntVector.class);
+ final IntVector key = exec.getValueVectorById(SchemaPath.getSimplePath("blue"), IntVector.class);
+ final long[] cntArr = {10001, 9999};
+ final int[] keyArr = {Integer.MIN_VALUE, Integer.MAX_VALUE};
- for(int i =0; i < exec.getRecordCount(); i++){
+ for(int i = 0; i < exec.getRecordCount(); i++) {
assertEquals((Long) cntArr[i], cnt.getAccessor().getObject(i));
assertEquals((Integer) keyArr[i], key.getAccessor().getObject(i));
}
}
- if(exec.getContext().getFailureCause() != null){
+ if(exec.getContext().getFailureCause() != null) {
throw exec.getContext().getFailureCause();
}
assertTrue(!exec.getContext().isFailed());
@@ -94,20 +95,20 @@ public class TestAgg extends ExecTest {
}
@Test
- public void twoKeyAgg(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable{
+ public void twoKeyAgg(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable {
SimpleRootExec exec = doTest(bitContext, connection, "/agg/twokey.json");
- while(exec.next()){
- IntVector key1 = exec.getValueVectorById(SchemaPath.getSimplePath("key1"), IntVector.class);
- BigIntVector key2 = exec.getValueVectorById(SchemaPath.getSimplePath("key2"), BigIntVector.class);
- BigIntVector cnt = exec.getValueVectorById(SchemaPath.getSimplePath("cnt"), BigIntVector.class);
- NullableBigIntVector total = exec.getValueVectorById(SchemaPath.getSimplePath("total"), NullableBigIntVector.class);
- Integer[] keyArr1 = {Integer.MIN_VALUE, Integer.MIN_VALUE, Integer.MIN_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE};
- long[] keyArr2 = {0,1,2,0,1,2};
- long[] cntArr = {34,34,34,34,34,34};
- long[] totalArr = {0,34,68,0,34,68};
-
- for(int i =0; i < exec.getRecordCount(); i++){
+ while(exec.next()) {
+ final IntVector key1 = exec.getValueVectorById(SchemaPath.getSimplePath("key1"), IntVector.class);
+ final BigIntVector key2 = exec.getValueVectorById(SchemaPath.getSimplePath("key2"), BigIntVector.class);
+ final BigIntVector cnt = exec.getValueVectorById(SchemaPath.getSimplePath("cnt"), BigIntVector.class);
+ final NullableBigIntVector total = exec.getValueVectorById(SchemaPath.getSimplePath("total"), NullableBigIntVector.class);
+ final Integer[] keyArr1 = {Integer.MIN_VALUE, Integer.MIN_VALUE, Integer.MIN_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE};
+ final long[] keyArr2 = {0,1,2,0,1,2};
+ final long[] cntArr = {34,34,34,34,34,34};
+ final long[] totalArr = {0,34,68,0,34,68};
+
+ for(int i = 0; i < exec.getRecordCount(); i++) {
// System.out.print(key1.getAccessor().getObject(i));
// System.out.print("\t");
// System.out.print(key2.getAccessor().getObject(i));
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/common/TestHashTable.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/common/TestHashTable.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/common/TestHashTable.java
index b02249d..84b652e 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/common/TestHashTable.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/common/TestHashTable.java
@@ -17,13 +17,20 @@
*/
package org.apache.drill.exec.physical.impl.common;
+import mockit.Injectable;
import mockit.NonStrictExpectations;
import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.expression.ExpressionPosition;
+import org.apache.drill.common.expression.FieldReference;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.logical.data.NamedExpression;
import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
+import org.apache.drill.exec.expr.holders.IntHolder;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.memory.TopLevelAllocator;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
@@ -36,56 +43,32 @@ import org.apache.drill.exec.proto.BitControl.PlanFragment;
import org.apache.drill.exec.proto.CoordinationProtos;
import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
import org.apache.drill.exec.server.DrillbitContext;
+import org.junit.Ignore;
+import org.junit.Test;
import com.codahale.metrics.MetricRegistry;
import com.google.common.base.Charsets;
import com.google.common.io.Files;
public class TestHashTable extends ExecTest {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestHashTable.class);
- DrillConfig c = DrillConfig.create();
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestHashTable.class);
+ private final DrillConfig c = DrillConfig.create();
@SuppressWarnings("deprecation")
private SimpleRootExec doTest(final DrillbitContext bitContext, UserClientConnection connection, String plan_path) throws Exception{
- new NonStrictExpectations(){{
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
-
-
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile(plan_path), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile(plan_path), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
return exec;
}
-/*
- @Test
- public void testHashTable1(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable {
- String plan_path = "/common/test_hashtable1.json";
- SimpleRootExec exec = doTest(bitContext, connection, plan_path);
-
- SchemaPath regionkey = new SchemaPath("regionkey", ExpressionPosition.UNKNOWN);
- SchemaPath nationkey = new SchemaPath("nationkey", ExpressionPosition.UNKNOWN);
-
- NamedExpression[] keyExprs = new NamedExpression[1];
- keyExprs[0] = new NamedExpression(regionkey, new FieldReference(regionkey));
-
- HashTableConfig htConfig = new HashTableConfig(HashTable.DEFAULT_INITIAL_CAPACITY, HashTable.DEFAULT_LOAD_FACTOR, keyExprs);
- ChainedHashTable cht = new ChainedHashTable(htConfig, exec.getContext(), exec.getIncoming(), null);
- HashTable htable = cht.createAndSetupHashTable();
- IntHolder htIdxHolder = new IntHolder();
-
- for (int i = 0; i < exec.getRecordCount(); i++) {
- HashTable.PutStatus putStatus = htable.put(i, htIdxHolder);
- assertNotEquals(putStatus, HashTable.PutStatus.PUT_FAILED);
- }
- assertEquals(htable.size(), 5);
- }
- */
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/filter/TestSimpleFilter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/filter/TestSimpleFilter.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/filter/TestSimpleFilter.java
index a069078..07a665a 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/filter/TestSimpleFilter.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/filter/TestSimpleFilter.java
@@ -27,7 +27,7 @@ import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -47,58 +47,53 @@ import com.google.common.base.Charsets;
import com.google.common.io.Files;
public class TestSimpleFilter extends ExecTest {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestSimpleFilter.class);
- DrillConfig c = DrillConfig.create();
-
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestSimpleFilter.class);
+ private final DrillConfig c = DrillConfig.create();
@Test
- public void testFilter(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable{
+ public void testFilter(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable {
// System.out.println(System.getProperty("java.class.path"));
-
-
- new NonStrictExpectations(){{
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
-
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/filter/test1.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- while(exec.next()){
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/filter/test1.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ while(exec.next()) {
assertEquals(50, exec.getRecordCount());
}
exec.close();
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
-
}
@Test
@Ignore ("Filter does not support SV4")
- public void testSV4Filter(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable{
- new NonStrictExpectations(){{
+ public void testSV4Filter(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable {
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/filter/test_sv4.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/filter/test_sv4.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
int recordCount = 0;
while(exec.next()) {
for (int i = 0; i < exec.getSelectionVector4().getCount(); i++) {
@@ -109,11 +104,9 @@ public class TestSimpleFilter extends ExecTest {
exec.close();
assertEquals(50, recordCount);
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
-
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoin.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoin.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoin.java
index 6c067fe..e7187d5 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoin.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoin.java
@@ -32,7 +32,7 @@ import org.apache.drill.common.util.TestTools;
import org.apache.drill.exec.client.DrillClient;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -63,11 +63,11 @@ import com.google.common.io.Files;
public class TestHashJoin extends PopUnitTestBase {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestMergeJoin.class);
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestMergeJoin.class);
@Rule public final TestRule TIMEOUT = TestTools.getTimeoutRule(100000);
- DrillConfig c = DrillConfig.create();
+ private final DrillConfig c = DrillConfig.create();
private void testHJMockScanCommon(final DrillbitContext bitContext, UserServer.UserClientConnection connection, String physicalPlan, int expectedRows) throws Throwable {
final LocalPStoreProvider provider = new LocalPStoreProvider(c);
@@ -76,18 +76,18 @@ public class TestHashJoin extends PopUnitTestBase {
opt.init();
new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getOptionManager(); result = opt;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile(physicalPlan), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile(physicalPlan), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
int totalRecordCount = 0;
while (exec.next()) {
@@ -169,18 +169,18 @@ public class TestHashJoin extends PopUnitTestBase {
@Injectable UserServer.UserClientConnection connection) throws Throwable {
// Function tests with hash join with exchanges
- try (RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- Drillbit bit = new Drillbit(CONFIG, serviceSet);
- DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator())) {
+ try (final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+ final Drillbit bit = new Drillbit(CONFIG, serviceSet);
+ final DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator())) {
// run query.
bit.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
Files.toString(FileUtils.getResourceAsFile("/join/hj_exchanges.json"), Charsets.UTF_8));
int count = 0;
- for (QueryDataBatch b : results) {
+ for (final QueryDataBatch b : results) {
if (b.getHeader().getRowCount() != 0) {
count += b.getHeader().getRowCount();
}
@@ -197,32 +197,32 @@ public class TestHashJoin extends PopUnitTestBase {
@Injectable UserServer.UserClientConnection connection) throws Throwable {
// Function tests hash join with multiple join conditions
- try (RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- Drillbit bit = new Drillbit(CONFIG, serviceSet);
- DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator())) {
+ try (final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+ final Drillbit bit = new Drillbit(CONFIG, serviceSet);
+ final DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator())) {
// run query.
bit.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
Files.toString(FileUtils.getResourceAsFile("/join/hj_multi_condition_join.json"), Charsets.UTF_8)
.replace("#{TEST_FILE_1}", FileUtils.getResourceAsFile("/build_side_input.json").toURI().toString())
.replace("#{TEST_FILE_2}", FileUtils.getResourceAsFile("/probe_side_input.json").toURI().toString()));
- RecordBatchLoader batchLoader = new RecordBatchLoader(bit.getContext().getAllocator());
+ final RecordBatchLoader batchLoader = new RecordBatchLoader(bit.getContext().getAllocator());
- QueryDataBatch batch = results.get(1);
+ final QueryDataBatch batch = results.get(1);
assertTrue(batchLoader.load(batch.getHeader().getDef(), batch.getData()));
- Iterator<VectorWrapper<?>> itr = batchLoader.iterator();
+ final Iterator<VectorWrapper<?>> itr = batchLoader.iterator();
// Just test the join key
- long colA[] = {1, 2, 1};
- long colC[] = {100, 200, 500};
+ final long colA[] = {1, 2, 1};
+ final long colC[] = {100, 200, 500};
// Check the output of decimal9
- ValueVector.Accessor intAccessor1 = itr.next().getValueVector().getAccessor();
- ValueVector.Accessor intAccessor2 = itr.next().getValueVector().getAccessor();
+ final ValueVector.Accessor intAccessor1 = itr.next().getValueVector().getAccessor();
+ final ValueVector.Accessor intAccessor2 = itr.next().getValueVector().getAccessor();
for (int i = 0; i < intAccessor1.getValueCount(); i++) {
@@ -232,30 +232,29 @@ public class TestHashJoin extends PopUnitTestBase {
assertEquals(3, intAccessor1.getValueCount());
batchLoader.clear();
- for (QueryDataBatch result : results) {
+ for (final QueryDataBatch result : results) {
result.release();
}
}
}
-
@Test
public void hjWithExchange1(@Injectable final DrillbitContext bitContext,
@Injectable UserServer.UserClientConnection connection) throws Throwable {
// Another test for hash join with exchanges
- try (RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- Drillbit bit = new Drillbit(CONFIG, serviceSet);
- DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator())) {
+ try (final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+ final Drillbit bit = new Drillbit(CONFIG, serviceSet);
+ final DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator())) {
// run query.
bit.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
Files.toString(FileUtils.getResourceAsFile("/join/hj_exchanges1.json"), Charsets.UTF_8));
int count = 0;
- for (QueryDataBatch b : results) {
+ for (final QueryDataBatch b : results) {
if (b.getHeader().getRowCount() != 0) {
count += b.getHeader().getRowCount();
}
@@ -267,20 +266,19 @@ public class TestHashJoin extends PopUnitTestBase {
}
}
-
@Test
public void testHashJoinExprInCondition() throws Exception {
- RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+ final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- try (Drillbit bit1 = new Drillbit(CONFIG, serviceSet);
- DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
+ try (final Drillbit bit1 = new Drillbit(CONFIG, serviceSet);
+ final DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
bit1.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
Files.toString(FileUtils.getResourceAsFile("/join/hashJoinExpr.json"), Charsets.UTF_8));
int count = 0;
- for (QueryDataBatch b : results) {
+ for (final QueryDataBatch b : results) {
if (b.getHeader().getRowCount() != 0) {
count += b.getHeader().getRowCount();
}
@@ -289,5 +287,4 @@ public class TestHashJoin extends PopUnitTestBase {
assertEquals(10, count);
}
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java
index 18555c7..58c55b6 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java
@@ -31,7 +31,7 @@ import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.client.DrillClient;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -60,9 +60,8 @@ import com.google.common.io.Files;
public class TestMergeJoin extends PopUnitTestBase {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestMergeJoin.class);
-
- DrillConfig c = DrillConfig.create();
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestMergeJoin.class);
+ private final DrillConfig c = DrillConfig.create();
@Test
@Ignore // this doesn't have a sort. it also causes an infinite loop. these may or may not be related.
@@ -71,36 +70,36 @@ public class TestMergeJoin extends PopUnitTestBase {
new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/join/merge_join.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/join/merge_join.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
int totalRecordCount = 0;
while (exec.next()) {
totalRecordCount += exec.getRecordCount();
- for (ValueVector v : exec) {
+ for (final ValueVector v : exec) {
System.out.print("[" + v.getField().toExpr() + "] ");
}
System.out.println("\n");
for (int valueIdx = 0; valueIdx < exec.getRecordCount(); valueIdx++) {
- List<Object> row = new ArrayList();
- for (ValueVector v : exec) {
+ final List<Object> row = new ArrayList();
+ for (final ValueVector v : exec) {
row.add(v.getAccessor().getObject(valueIdx));
}
- for (Object cell : row) {
+ for (final Object cell : row) {
if (cell == null) {
System.out.print("<null> ");
continue;
}
- int len = cell.toString().length();
+ final int len = cell.toString().length();
System.out.print(cell);
for (int i = 0; i < (14 - len); ++i) {
System.out.print(" ");
@@ -116,31 +115,29 @@ public class TestMergeJoin extends PopUnitTestBase {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
-
}
@Test
@Ignore
public void orderedEqualityLeftJoin(@Injectable final DrillbitContext bitContext,
@Injectable UserServer.UserClientConnection connection) throws Throwable {
-
new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getConfig(); result = c;
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(),CoordinationProtos.DrillbitEndpoint.getDefaultInstance(), new StoragePluginRegistry(bitContext));
- PhysicalPlan plan = reader.readPhysicalPlan(
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(),CoordinationProtos.DrillbitEndpoint.getDefaultInstance(), new StoragePluginRegistry(bitContext));
+ final PhysicalPlan plan = reader.readPhysicalPlan(
Files.toString(
FileUtils.getResourceAsFile("/join/merge_single_batch.json"), Charsets.UTF_8)
.replace("#{LEFT_FILE}", FileUtils.getResourceAsFile("/join/merge_single_batch.left.json").toURI().toString())
.replace("#{RIGHT_FILE}", FileUtils.getResourceAsFile("/join/merge_single_batch.right.json").toURI().toString()));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
int totalRecordCount = 0;
while (exec.next()) {
@@ -149,16 +146,16 @@ public class TestMergeJoin extends PopUnitTestBase {
System.out.println(" t1 t2");
for (int valueIdx = 0; valueIdx < exec.getRecordCount(); valueIdx++) {
- List<Object> row = Lists.newArrayList();
- for (ValueVector v : exec) {
+ final List<Object> row = Lists.newArrayList();
+ for (final ValueVector v : exec) {
row.add(v.getField().toExpr() + ":" + v.getAccessor().getObject(valueIdx));
}
- for (Object cell : row) {
+ for (final Object cell : row) {
if (cell == null) {
System.out.print("<null> ");
continue;
}
- int len = cell.toString().length();
+ final int len = cell.toString().length();
System.out.print(cell + " ");
for (int i = 0; i < (10 - len); ++i) {
System.out.print(" ");
@@ -174,31 +171,29 @@ public class TestMergeJoin extends PopUnitTestBase {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
-
}
@Test
@Ignore
public void orderedEqualityInnerJoin(@Injectable final DrillbitContext bitContext,
@Injectable UserServer.UserClientConnection connection) throws Throwable {
-
new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getConfig(); result = c;
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(),CoordinationProtos.DrillbitEndpoint.getDefaultInstance(), new StoragePluginRegistry(bitContext));
- PhysicalPlan plan = reader.readPhysicalPlan(
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(),CoordinationProtos.DrillbitEndpoint.getDefaultInstance(), new StoragePluginRegistry(bitContext));
+ final PhysicalPlan plan = reader.readPhysicalPlan(
Files.toString(
FileUtils.getResourceAsFile("/join/merge_inner_single_batch.json"), Charsets.UTF_8)
.replace("#{LEFT_FILE}", FileUtils.getResourceAsFile("/join/merge_single_batch.left.json").toURI().toString())
.replace("#{RIGHT_FILE}", FileUtils.getResourceAsFile("/join/merge_single_batch.right.json").toURI().toString()));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
int totalRecordCount = 0;
while (exec.next()) {
@@ -207,16 +202,16 @@ public class TestMergeJoin extends PopUnitTestBase {
System.out.println(" t1 t2");
for (int valueIdx = 0; valueIdx < exec.getRecordCount(); valueIdx++) {
- List<Object> row = Lists.newArrayList();
- for (ValueVector v : exec) {
+ final List<Object> row = Lists.newArrayList();
+ for (final ValueVector v : exec) {
row.add(v.getField().toExpr() + ":" + v.getAccessor().getObject(valueIdx));
}
- for (Object cell : row) {
+ for (final Object cell : row) {
if (cell == null) {
System.out.print("<null> ");
continue;
}
- int len = cell.toString().length();
+ final int len = cell.toString().length();
System.out.print(cell + " ");
for (int i = 0; i < (10 - len); ++i) {
System.out.print(" ");
@@ -232,31 +227,29 @@ public class TestMergeJoin extends PopUnitTestBase {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
-
}
@Test
@Ignore
public void orderedEqualityMultiBatchJoin(@Injectable final DrillbitContext bitContext,
@Injectable UserServer.UserClientConnection connection) throws Throwable {
-
new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getConfig(); result = c;
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(),CoordinationProtos.DrillbitEndpoint.getDefaultInstance(), new StoragePluginRegistry(bitContext));
- PhysicalPlan plan = reader.readPhysicalPlan(
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(),CoordinationProtos.DrillbitEndpoint.getDefaultInstance(), new StoragePluginRegistry(bitContext));
+ final PhysicalPlan plan = reader.readPhysicalPlan(
Files.toString(
FileUtils.getResourceAsFile("/join/merge_multi_batch.json"), Charsets.UTF_8)
.replace("#{LEFT_FILE}", FileUtils.getResourceAsFile("/join/merge_multi_batch.left.json").toURI().toString())
.replace("#{RIGHT_FILE}", FileUtils.getResourceAsFile("/join/merge_multi_batch.right.json").toURI().toString()));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
int totalRecordCount = 0;
while (exec.next()) {
@@ -264,11 +257,11 @@ public class TestMergeJoin extends PopUnitTestBase {
System.out.println("got next with record count: " + exec.getRecordCount() + " (total: " + totalRecordCount + "):");
for (int valueIdx = 0; valueIdx < exec.getRecordCount(); valueIdx++) {
- List<Object> row = Lists.newArrayList();
- for (ValueVector v : exec) {
+ final List<Object> row = Lists.newArrayList();
+ for (final ValueVector v : exec) {
row.add(v.getField().toExpr() + ":" + v.getAccessor().getObject(valueIdx));
}
- for (Object cell : row) {
+ for (final Object cell : row) {
if (cell == null) {
System.out.print("<null> ");
continue;
@@ -289,24 +282,23 @@ public class TestMergeJoin extends PopUnitTestBase {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
-
}
@Test
public void testJoinBatchSize(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable{
new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();;
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getConfig(); result = c;
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/join/join_batchsize.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/join/join_batchsize.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
exec.next(); // skip schema batch
while (exec.next()) {
assertEquals(100, exec.getRecordCount());
@@ -316,24 +308,23 @@ public class TestMergeJoin extends PopUnitTestBase {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
-
}
@Test
public void testMergeJoinInnerEmptyBatch() throws Exception {
- RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+ final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- try (Drillbit bit1 = new Drillbit(CONFIG, serviceSet);
- DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
+ try (final Drillbit bit1 = new Drillbit(CONFIG, serviceSet);
+ final DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
bit1.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
Files.toString(FileUtils.getResourceAsFile("/join/merge_join_empty_batch.json"),
Charsets.UTF_8)
.replace("${JOIN_TYPE}", "INNER"));
int count = 0;
- for (QueryDataBatch b : results) {
+ for (final QueryDataBatch b : results) {
if (b.getHeader().getRowCount() != 0) {
count += b.getHeader().getRowCount();
}
@@ -345,19 +336,19 @@ public class TestMergeJoin extends PopUnitTestBase {
@Test
public void testMergeJoinLeftEmptyBatch() throws Exception {
- RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+ final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- try (Drillbit bit1 = new Drillbit(CONFIG, serviceSet);
- DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
+ try (final Drillbit bit1 = new Drillbit(CONFIG, serviceSet);
+ final DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
bit1.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
Files.toString(FileUtils.getResourceAsFile("/join/merge_join_empty_batch.json"),
Charsets.UTF_8)
.replace("${JOIN_TYPE}", "LEFT"));
int count = 0;
- for (QueryDataBatch b : results) {
+ for (final QueryDataBatch b : results) {
if (b.getHeader().getRowCount() != 0) {
count += b.getHeader().getRowCount();
}
@@ -369,19 +360,19 @@ public class TestMergeJoin extends PopUnitTestBase {
@Test
public void testMergeJoinRightEmptyBatch() throws Exception {
- RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+ final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- try (Drillbit bit1 = new Drillbit(CONFIG, serviceSet);
- DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
+ try (final Drillbit bit1 = new Drillbit(CONFIG, serviceSet);
+ final DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
bit1.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
Files.toString(FileUtils.getResourceAsFile("/join/merge_join_empty_batch.json"),
Charsets.UTF_8)
.replace("${JOIN_TYPE}", "RIGHT"));
int count = 0;
- for (QueryDataBatch b : results) {
+ for (final QueryDataBatch b : results) {
if (b.getHeader().getRowCount() != 0) {
count += b.getHeader().getRowCount();
}
@@ -393,17 +384,17 @@ public class TestMergeJoin extends PopUnitTestBase {
@Test
public void testMergeJoinExprInCondition() throws Exception {
- RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+ final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- try (Drillbit bit1 = new Drillbit(CONFIG, serviceSet);
- DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
+ try (final Drillbit bit1 = new Drillbit(CONFIG, serviceSet);
+ final DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
bit1.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
Files.toString(FileUtils.getResourceAsFile("/join/mergeJoinExpr.json"), Charsets.UTF_8));
int count = 0;
- for (QueryDataBatch b : results) {
+ for (final QueryDataBatch b : results) {
if (b.getHeader().getRowCount() != 0) {
count += b.getHeader().getRowCount();
}
@@ -412,5 +403,4 @@ public class TestMergeJoin extends PopUnitTestBase {
assertEquals(10, count);
}
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/limit/TestSimpleLimit.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/limit/TestSimpleLimit.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/limit/TestSimpleLimit.java
index 7cdb41a..97bbd2c 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/limit/TestSimpleLimit.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/limit/TestSimpleLimit.java
@@ -27,7 +27,7 @@ import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -48,13 +48,13 @@ import com.google.common.base.Charsets;
import com.google.common.io.Files;
public class TestSimpleLimit extends ExecTest {
- DrillConfig c = DrillConfig.create();
+ private final DrillConfig c = DrillConfig.create();
@Test
- public void testLimit(@Injectable final DrillbitContext bitContext, @Injectable UserServer.UserClientConnection connection) throws Throwable{
- new NonStrictExpectations(){{
+ public void testLimit(@Injectable final DrillbitContext bitContext, @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
@@ -64,10 +64,10 @@ public class TestSimpleLimit extends ExecTest {
}
@Test
- public void testLimitNoEnd(@Injectable final DrillbitContext bitContext, @Injectable UserServer.UserClientConnection connection) throws Throwable{
- new NonStrictExpectations(){{
+ public void testLimitNoEnd(@Injectable final DrillbitContext bitContext, @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
@@ -82,19 +82,19 @@ public class TestSimpleLimit extends ExecTest {
// However, when evaluate the increasingBitInt(0), if the outgoing batch could not hold the new value, doEval() return false, and start the
// next batch. But the value has already been increased by 1 in the prior failed try. Therefore, the sum of the generated number could be different,
// depending on the size of each outgoing batch, and when the batch could not hold any more values.
- public void testLimitAcrossBatches(@Injectable final DrillbitContext bitContext, @Injectable UserServer.UserClientConnection connection) throws Throwable{
+ public void testLimitAcrossBatches(@Injectable final DrillbitContext bitContext, @Injectable UserServer.UserClientConnection connection) throws Throwable {
new NonStrictExpectations(){{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
verifyLimitCount(bitContext, connection, "test2.json", 69999);
- long start = 30000;
- long end = 100000;
- long expectedSum = (end - start) * (end + start - 1) / 2; //Formula for sum of series
+ final long start = 30000;
+ final long end = 100000;
+ final long expectedSum = (end - start) * (end + start - 1) / 2; //Formula for sum of series
verifySum(bitContext, connection, "test4.json", 70000, expectedSum);
@@ -102,19 +102,19 @@ public class TestSimpleLimit extends ExecTest {
}
private void verifyLimitCount(DrillbitContext bitContext, UserServer.UserClientConnection connection, String testPlan, int expectedCount) throws Throwable {
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/limit/" + testPlan), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/limit/" + testPlan), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
int recordCount = 0;
- while(exec.next()){
+ while(exec.next()) {
recordCount += exec.getRecordCount();
}
assertEquals(expectedCount, recordCount);
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
@@ -122,16 +122,16 @@ public class TestSimpleLimit extends ExecTest {
}
private void verifySum(DrillbitContext bitContext, UserServer.UserClientConnection connection, String testPlan, int expectedCount, long expectedSum) throws Throwable {
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/limit/" + testPlan), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/limit/" + testPlan), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
int recordCount = 0;
long sum = 0;
- while(exec.next()){
+ while(exec.next()) {
recordCount += exec.getRecordCount();
- BigIntVector v = (BigIntVector) exec.iterator().next();
+ final BigIntVector v = (BigIntVector) exec.iterator().next();
for (int i = 0; i < v.getAccessor().getValueCount(); i++) {
sum += v.getAccessor().get(i);
}
@@ -140,7 +140,7 @@ public class TestSimpleLimit extends ExecTest {
assertEquals(expectedCount, recordCount);
assertEquals(expectedSum, sum);
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/project/TestSimpleProjection.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/project/TestSimpleProjection.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/project/TestSimpleProjection.java
index 43c430a..e8cf7fe 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/project/TestSimpleProjection.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/project/TestSimpleProjection.java
@@ -29,6 +29,7 @@ import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.memory.TopLevelAllocator;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
@@ -50,34 +51,32 @@ import com.google.common.base.Charsets;
import com.google.common.io.Files;
public class TestSimpleProjection extends ExecTest {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestSimpleProjection.class);
- DrillConfig c = DrillConfig.create();
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestSimpleProjection.class);
+ private final DrillConfig c = DrillConfig.create();
@Test
- public void project(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable{
-
+ public void project(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable {
new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/project/test1.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/project/test1.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
while (exec.next()) {
VectorUtil.showVectorAccessibleContent(exec.getIncoming(), "\t");
- NullableBigIntVector c1 = exec.getValueVectorById(new SchemaPath("col1", ExpressionPosition.UNKNOWN), NullableBigIntVector.class);
- NullableBigIntVector c2 = exec.getValueVectorById(new SchemaPath("col2", ExpressionPosition.UNKNOWN), NullableBigIntVector.class);
+ final NullableBigIntVector c1 = exec.getValueVectorById(new SchemaPath("col1", ExpressionPosition.UNKNOWN), NullableBigIntVector.class);
+ final NullableBigIntVector c2 = exec.getValueVectorById(new SchemaPath("col2", ExpressionPosition.UNKNOWN), NullableBigIntVector.class);
int x = 0;
- NullableBigIntVector.Accessor a1, a2;
- a1 = c1.getAccessor();
- a2 = c2.getAccessor();
+ final NullableBigIntVector.Accessor a1 = c1.getAccessor();
+ final NullableBigIntVector.Accessor a2 = c2.getAccessor();
for (int i =0; i < c1.getAccessor().getValueCount(); i++) {
if (!a1.isNull(i)) {
@@ -92,5 +91,4 @@ public class TestSimpleProjection extends ExecTest {
}
assertTrue(!context.isFailed());
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/sort/TestSimpleSort.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/sort/TestSimpleSort.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/sort/TestSimpleSort.java
index d51a017..4556eb7 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/sort/TestSimpleSort.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/sort/TestSimpleSort.java
@@ -29,7 +29,7 @@ import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -56,10 +56,10 @@ public class TestSimpleSort extends ExecTest {
private final DrillConfig c = DrillConfig.create();
@Test
- public void sortOneKeyAscending(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable{
- new NonStrictExpectations(){{
+ public void sortOneKeyAscending(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable {
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
@@ -100,10 +100,10 @@ public class TestSimpleSort extends ExecTest {
}
@Test
- public void sortTwoKeysOneAscendingOneDescending(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable{
+ public void sortTwoKeysOneAscendingOneDescending(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable {
new NonStrictExpectations(){{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
@@ -113,7 +113,7 @@ public class TestSimpleSort extends ExecTest {
final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/sort/two_key_sort.json"), Charsets.UTF_8));
final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
int previousInt = Integer.MIN_VALUE;
long previousLong = Long.MAX_VALUE;
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/trace/TestTraceMultiRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/trace/TestTraceMultiRecordBatch.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/trace/TestTraceMultiRecordBatch.java
index b82846e..a02f3b5 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/trace/TestTraceMultiRecordBatch.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/trace/TestTraceMultiRecordBatch.java
@@ -26,6 +26,7 @@ import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.memory.TopLevelAllocator;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
@@ -54,39 +55,36 @@ import com.google.common.io.Files;
* incoming container of the trace operator.
*/
public class TestTraceMultiRecordBatch extends ExecTest {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestTraceOutputDump.class);
- DrillConfig c = DrillConfig.create();
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestTraceOutputDump.class);
+ private final DrillConfig c = DrillConfig.create();
+ @Test
+ public void testFilter(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable {
+ new NonStrictExpectations() {{
+ bitContext.getMetrics(); result = new MetricRegistry();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
+ bitContext.getConfig(); result = c;
+ bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
+ bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
+ }};
- @Test
- public void testFilter(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable
- {
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/trace/multi_record_batch_trace.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- new NonStrictExpectations(){{
- bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
- bitContext.getConfig(); result = c;
- bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
- bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
- }};
-
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/trace/multi_record_batch_trace.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
-
- while(exec.next()) {
- for(ValueVector vv: exec){
- vv.clear();
- }
- }
+ while(exec.next()) {
+ for(final ValueVector vv: exec){
+ vv.clear();
+ }
+ }
- exec.close();
+ exec.close();
- if(context.getFailureCause() != null){
- throw context.getFailureCause();
- }
- assertTrue(!context.isFailed());
+ if(context.getFailureCause() != null) {
+ throw context.getFailureCause();
}
-}
\ No newline at end of file
+ assertTrue(!context.isFailed());
+ }
+}
[5/5] drill git commit: DRILL-3669: Fix missing indirect dependency.
Posted by js...@apache.org.
DRILL-3669: Fix missing indirect dependency.
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/4b8e85ad
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/4b8e85ad
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/4b8e85ad
Branch: refs/heads/master
Commit: 4b8e85ad6fb40554e6752144f09bdfb474d62d9b
Parents: bfe6cfa
Author: Julien Le Dem <ju...@ledem.net>
Authored: Tue Aug 18 22:24:52 2015 -0700
Committer: Jason Altekruse <al...@gmail.com>
Committed: Wed Sep 2 14:06:24 2015 -0700
----------------------------------------------------------------------
exec/java-exec/pom.xml | 6 ++++++
1 file changed, 6 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/4b8e85ad/exec/java-exec/pom.xml
----------------------------------------------------------------------
diff --git a/exec/java-exec/pom.xml b/exec/java-exec/pom.xml
index 601f5ca..cb48567 100644
--- a/exec/java-exec/pom.xml
+++ b/exec/java-exec/pom.xml
@@ -21,6 +21,12 @@
<name>exec/Java Execution Engine</name>
<dependencies>
+ <dependency>
+ <groupId>org.hamcrest</groupId>
+ <artifactId>hamcrest-core</artifactId>
+ <version>1.3</version>
+ <scope>test</scope>
+ </dependency>
<!-- <dependency> -->
<!-- <groupId>org.ow2.asm</groupId> -->
<!-- <artifactId>asm-util</artifactId> -->
[3/5] drill git commit: DRILL-3598: use a factory to create the root
allocator. - made the constructor for TopLevelAllocator package private to
enforce this
Posted by js...@apache.org.
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestCastFunctions.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestCastFunctions.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestCastFunctions.java
index ffa8765..7e1da44 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestCastFunctions.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestCastFunctions.java
@@ -40,7 +40,7 @@ import org.apache.drill.exec.expr.holders.IntHolder;
import org.apache.drill.exec.expr.holders.VarBinaryHolder;
import org.apache.drill.exec.expr.holders.VarCharHolder;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -69,19 +69,16 @@ import com.google.common.base.Charsets;
import com.google.common.io.Files;
public class TestCastFunctions extends PopUnitTestBase{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestSimpleFunctions.class);
-
- DrillConfig c = DrillConfig.create();
-
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestSimpleFunctions.class);
+ private final DrillConfig c = DrillConfig.create();
@Test
// cast to bigint.
public void testCastBigInt(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
-
- final BufferAllocator allocator = new TopLevelAllocator();
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final BufferAllocator allocator = RootAllocatorFactory.newRoot(c);
- new NonStrictExpectations(){{
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
bitContext.getAllocator(); result = allocator;
bitContext.getConfig(); result = c;
@@ -89,19 +86,18 @@ public class TestCastFunctions extends PopUnitTestBase{
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastBigInt.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastBigInt.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- while(exec.next()){
- BigIntVector c0 = exec.getValueVectorById(new SchemaPath("varchar_cast", ExpressionPosition.UNKNOWN), BigIntVector.class);
- BigIntVector.Accessor a0;
- a0 = c0.getAccessor();
+ while(exec.next()) {
+ final BigIntVector c0 = exec.getValueVectorById(new SchemaPath("varchar_cast", ExpressionPosition.UNKNOWN), BigIntVector.class);
+ final BigIntVector.Accessor a0 = c0.getAccessor();
int count = 0;
- for(int i = 0; i < c0.getAccessor().getValueCount(); i++){
+ for(int i = 0; i < c0.getAccessor().getValueCount(); i++) {
BigIntHolder holder0 = new BigIntHolder();
a0.get(i, holder0);
assertEquals(1256, holder0.value);
@@ -116,7 +112,7 @@ public class TestCastFunctions extends PopUnitTestBase{
context.close();
allocator.close();
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
@@ -125,11 +121,11 @@ public class TestCastFunctions extends PopUnitTestBase{
@Test
//cast to int
public void testCastInt(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
- final BufferAllocator allocator = new TopLevelAllocator();
+ final BufferAllocator allocator = RootAllocatorFactory.newRoot(c);
- new NonStrictExpectations(){{
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
bitContext.getAllocator(); result = allocator;
bitContext.getConfig(); result = c;
@@ -137,24 +133,22 @@ public class TestCastFunctions extends PopUnitTestBase{
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastInt.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastInt.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- while(exec.next()){
- IntVector c0 = exec.getValueVectorById(new SchemaPath("varchar_cast", ExpressionPosition.UNKNOWN), IntVector.class);
- IntVector.Accessor a0;
- a0 = c0.getAccessor();
+ while(exec.next()) {
+ final IntVector c0 = exec.getValueVectorById(new SchemaPath("varchar_cast", ExpressionPosition.UNKNOWN), IntVector.class);
+ final IntVector.Accessor a0 = c0.getAccessor();
int count = 0;
- for(int i = 0; i < c0.getAccessor().getValueCount(); i++){
- IntHolder holder0 = new IntHolder();
+ for(int i = 0; i < c0.getAccessor().getValueCount(); i++) {
+ final IntHolder holder0 = new IntHolder();
a0.get(i, holder0);
assertEquals(1256, holder0.value);
++count;
-
}
assertEquals(5, count);
}
@@ -164,7 +158,7 @@ public class TestCastFunctions extends PopUnitTestBase{
context.close();
allocator.close();
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
@@ -173,9 +167,9 @@ public class TestCastFunctions extends PopUnitTestBase{
@Test
//cast to float4
public void testCastFloat4(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
- final BufferAllocator allocator = new TopLevelAllocator();
- new NonStrictExpectations(){{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final BufferAllocator allocator = RootAllocatorFactory.newRoot(c);
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
bitContext.getAllocator(); result = allocator;
bitContext.getConfig(); result = c;
@@ -183,20 +177,19 @@ public class TestCastFunctions extends PopUnitTestBase{
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastFloat4.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastFloat4.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- while(exec.next()){
- Float4Vector c0 = exec.getValueVectorById(new SchemaPath("varchar_cast2", ExpressionPosition.UNKNOWN), Float4Vector.class);
- Float4Vector.Accessor a0;
- a0 = c0.getAccessor();
+ while(exec.next()) {
+ final Float4Vector c0 = exec.getValueVectorById(new SchemaPath("varchar_cast2", ExpressionPosition.UNKNOWN), Float4Vector.class);
+ final Float4Vector.Accessor a0 = c0.getAccessor();
int count = 0;
- for(int i = 0; i < c0.getAccessor().getValueCount(); i++){
- Float4Holder holder0 = new Float4Holder();
+ for(int i = 0; i < c0.getAccessor().getValueCount(); i++) {
+ final Float4Holder holder0 = new Float4Holder();
a0.get(i, holder0);
assertEquals(12.56, holder0.value, 0.001);
++count;
@@ -210,7 +203,7 @@ public class TestCastFunctions extends PopUnitTestBase{
context.close();
allocator.close();
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
@@ -219,10 +212,9 @@ public class TestCastFunctions extends PopUnitTestBase{
@Test
//cast to float8
public void testCastFloat8(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
-
- final BufferAllocator allocator = new TopLevelAllocator();
- new NonStrictExpectations(){{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final BufferAllocator allocator = RootAllocatorFactory.newRoot(c);
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
bitContext.getAllocator(); result = allocator;
bitContext.getConfig(); result = c;
@@ -230,20 +222,19 @@ public class TestCastFunctions extends PopUnitTestBase{
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastFloat8.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastFloat8.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- while(exec.next()){
- Float8Vector c0 = exec.getValueVectorById(new SchemaPath("varchar_cast2", ExpressionPosition.UNKNOWN), Float8Vector.class);
- Float8Vector.Accessor a0;
- a0 = c0.getAccessor();
+ while(exec.next()) {
+ final Float8Vector c0 = exec.getValueVectorById(new SchemaPath("varchar_cast2", ExpressionPosition.UNKNOWN), Float8Vector.class);
+ final Float8Vector.Accessor a0 = c0.getAccessor();
int count = 0;
for(int i = 0; i < c0.getAccessor().getValueCount(); i++){
- Float8Holder holder0 = new Float8Holder();
+ final Float8Holder holder0 = new Float8Holder();
a0.get(i, holder0);
assertEquals(12.56, holder0.value, 0.001);
++count;
@@ -257,7 +248,7 @@ public class TestCastFunctions extends PopUnitTestBase{
context.close();
allocator.close();
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
@@ -266,10 +257,10 @@ public class TestCastFunctions extends PopUnitTestBase{
@Test
//cast to varchar(length)
public void testCastVarChar(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
- final BufferAllocator allocator = new TopLevelAllocator();
- new NonStrictExpectations(){{
+ final BufferAllocator allocator = RootAllocatorFactory.newRoot(c);
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
bitContext.getAllocator(); result = allocator;
bitContext.getConfig(); result = c;
@@ -277,24 +268,22 @@ public class TestCastFunctions extends PopUnitTestBase{
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastVarChar.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastVarChar.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- while(exec.next()){
- VarCharVector c0 = exec.getValueVectorById(new SchemaPath("int_lit_cast", ExpressionPosition.UNKNOWN), VarCharVector.class);
- VarCharVector.Accessor a0;
- a0 = c0.getAccessor();
+ while(exec.next()) {
+ final VarCharVector c0 = exec.getValueVectorById(new SchemaPath("int_lit_cast", ExpressionPosition.UNKNOWN), VarCharVector.class);
+ final VarCharVector.Accessor a0 = c0.getAccessor();
int count = 0;
- for(int i = 0; i < c0.getAccessor().getValueCount(); i++){
- VarCharHolder holder0 = new VarCharHolder();
+ for(int i = 0; i < c0.getAccessor().getValueCount(); i++) {
+ final VarCharHolder holder0 = new VarCharHolder();
a0.get(i, holder0);
assertEquals("123", StringFunctionHelpers.toStringFromUTF8(holder0.start, holder0.end, holder0.buffer));
++count;
-
}
assertEquals(5, count);
}
@@ -304,7 +293,7 @@ public class TestCastFunctions extends PopUnitTestBase{
context.close();
allocator.close();
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
@@ -313,11 +302,10 @@ public class TestCastFunctions extends PopUnitTestBase{
@Test
//cast to varbinary(length)
public void testCastVarBinary(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
-
- final BufferAllocator allocator = new TopLevelAllocator();
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final BufferAllocator allocator = RootAllocatorFactory.newRoot(c);
- new NonStrictExpectations(){{
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
bitContext.getAllocator(); result = allocator;
bitContext.getConfig(); result = c;
@@ -325,20 +313,19 @@ public class TestCastFunctions extends PopUnitTestBase{
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastVarBinary.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastVarBinary.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- while(exec.next()){
- VarBinaryVector c0 = exec.getValueVectorById(new SchemaPath("int_lit_cast", ExpressionPosition.UNKNOWN), VarBinaryVector.class);
- VarBinaryVector.Accessor a0;
- a0 = c0.getAccessor();
+ while(exec.next()) {
+ final VarBinaryVector c0 = exec.getValueVectorById(new SchemaPath("int_lit_cast", ExpressionPosition.UNKNOWN), VarBinaryVector.class);
+ final VarBinaryVector.Accessor a0 = c0.getAccessor();
int count = 0;
- for(int i = 0; i < c0.getAccessor().getValueCount(); i++){
- VarBinaryHolder holder0 = new VarBinaryHolder();
+ for(int i = 0; i < c0.getAccessor().getValueCount(); i++) {
+ final VarBinaryHolder holder0 = new VarBinaryHolder();
a0.get(i, holder0);
assertEquals("123", StringFunctionHelpers.toStringFromUTF8(holder0.start, holder0.end, holder0.buffer));
++count;
@@ -351,7 +338,7 @@ public class TestCastFunctions extends PopUnitTestBase{
context.close();
allocator.close();
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
@@ -360,10 +347,10 @@ public class TestCastFunctions extends PopUnitTestBase{
@Test
//nested: cast is nested in another cast, or another function.
public void testCastNested(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
- final BufferAllocator allocator = new TopLevelAllocator();
- new NonStrictExpectations(){{
+ final BufferAllocator allocator = RootAllocatorFactory.newRoot(c);
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
bitContext.getAllocator(); result = allocator;
bitContext.getConfig(); result = c;
@@ -371,20 +358,19 @@ public class TestCastFunctions extends PopUnitTestBase{
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastNested.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastNested.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- while(exec.next()){
- IntVector c0 = exec.getValueVectorById(new SchemaPath("add_cast", ExpressionPosition.UNKNOWN),IntVector.class);
- IntVector.Accessor a0;
- a0 = c0.getAccessor();
+ while(exec.next()) {
+ final IntVector c0 = exec.getValueVectorById(new SchemaPath("add_cast", ExpressionPosition.UNKNOWN),IntVector.class);
+ final IntVector.Accessor a0 = c0.getAccessor();
int count = 0;
- for(int i = 0; i < c0.getAccessor().getValueCount(); i++){
- IntHolder holder0 = new IntHolder();
+ for(int i = 0; i < c0.getAccessor().getValueCount(); i++) {
+ final IntHolder holder0 = new IntHolder();
a0.get(i, holder0);
assertEquals(300, holder0.value);
++count;
@@ -397,7 +383,7 @@ public class TestCastFunctions extends PopUnitTestBase{
context.close();
allocator.close();
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
@@ -407,11 +393,10 @@ public class TestCastFunctions extends PopUnitTestBase{
@Test(expected = NumberFormatException.class)
public void testCastNumException(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
-
- final BufferAllocator allocator = new TopLevelAllocator();
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ final BufferAllocator allocator = RootAllocatorFactory.newRoot(c);
- new NonStrictExpectations(){{
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
bitContext.getAllocator(); result = allocator;
bitContext.getConfig(); result = c;
@@ -419,13 +404,13 @@ public class TestCastFunctions extends PopUnitTestBase{
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastNumException.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastNumException.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- while(exec.next()){
+ while(exec.next()) {
}
exec.close();
@@ -435,32 +420,30 @@ public class TestCastFunctions extends PopUnitTestBase{
assertTrue(context.isFailed());
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
-
}
@Test
public void testCastFromNullablCol() throws Throwable {
- RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
-
- try(Drillbit bit = new Drillbit(CONFIG, serviceSet);
- DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator())) {
+ final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+ try(final Drillbit bit = new Drillbit(CONFIG, serviceSet);
+ final DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator())) {
bit.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
Files.toString(FileUtils.getResourceAsFile("/functions/cast/testCastVarCharNull.json"), Charsets.UTF_8).replace("#{TEST_FILE}", "/jsoninput/input1.json"));
- QueryDataBatch batch = results.get(0);
+ final QueryDataBatch batch = results.get(0);
- RecordBatchLoader batchLoader = new RecordBatchLoader(bit.getContext().getAllocator());
+ final RecordBatchLoader batchLoader = new RecordBatchLoader(bit.getContext().getAllocator());
batchLoader.load(batch.getHeader().getDef(), batch.getData());
- Object [][] result = getRunResult(batchLoader);
+ final Object [][] result = getRunResult(batchLoader);
- Object [][] expected = new Object[2][2];
+ final Object [][] expected = new Object[2][2];
expected[0][0] = new String("2001");
expected[0][1] = new String("1.2");
@@ -477,24 +460,23 @@ public class TestCastFunctions extends PopUnitTestBase{
}
}
batchLoader.clear();
- for(QueryDataBatch b : results){
+ for(final QueryDataBatch b : results){
b.release();
}
-
}
}
private Object[][] getRunResult(VectorAccessible va) {
int size = 0;
- for (VectorWrapper v : va) {
+ for (final VectorWrapper v : va) {
size++;
}
- Object[][] res = new Object [va.getRecordCount()][size];
+ final Object[][] res = new Object [va.getRecordCount()][size];
for (int j = 0; j < va.getRecordCount(); j++) {
int i = 0;
- for (VectorWrapper v : va) {
- Object o = v.getValueVector().getAccessor().getObject(j);
+ for (final VectorWrapper v : va) {
+ final Object o = v.getValueVector().getAccessor().getObject(j);
if (o instanceof byte[]) {
res[j][i++] = new String((byte[]) o);
} else {
@@ -504,5 +486,4 @@ public class TestCastFunctions extends PopUnitTestBase{
}
return res;
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestComparisonFunctions.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestComparisonFunctions.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestComparisonFunctions.java
index c69c6f5..eb2e3c0 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestComparisonFunctions.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestComparisonFunctions.java
@@ -26,7 +26,7 @@ import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -51,16 +51,15 @@ public class TestComparisonFunctions extends ExecTest {
public void runTest(@Injectable final DrillbitContext bitContext,
@Injectable UserServer.UserClientConnection connection, String expression, int expectedResults) throws Throwable {
-
new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- String planString = Resources.toString(Resources.getResource(COMPARISON_TEST_PHYSICAL_PLAN), Charsets.UTF_8).replaceAll("EXPRESSION", expression);
+ final String planString = Resources.toString(Resources.getResource(COMPARISON_TEST_PHYSICAL_PLAN), Charsets.UTF_8).replaceAll("EXPRESSION", expression);
if (reader == null) {
reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
}
@@ -69,8 +68,8 @@ public class TestComparisonFunctions extends ExecTest {
}
final FragmentContext context =
new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- PhysicalPlan plan = reader.readPhysicalPlan(planString);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlan plan = reader.readPhysicalPlan(planString);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
while(exec.next()) {
assertEquals(String.format("Expression: %s;", expression), expectedResults,
@@ -92,7 +91,7 @@ public class TestComparisonFunctions extends ExecTest {
@Test
public void testInt(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
runTest(bitContext, connection, "intColumn == intColumn", 100);
runTest(bitContext, connection, "intColumn != intColumn", 0);
runTest(bitContext, connection, "intColumn > intColumn", 0);
@@ -103,7 +102,7 @@ public class TestComparisonFunctions extends ExecTest {
@Test
public void testBigInt(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
runTest(bitContext, connection, "bigIntColumn == bigIntColumn", 100);
runTest(bitContext, connection, "bigIntColumn != bigIntColumn", 0);
runTest(bitContext, connection, "bigIntColumn > bigIntColumn", 0);
@@ -114,7 +113,7 @@ public class TestComparisonFunctions extends ExecTest {
@Test
public void testFloat4(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
runTest(bitContext, connection, "float4Column == float4Column", 100);
runTest(bitContext, connection, "float4Column != float4Column", 0);
runTest(bitContext, connection, "float4Column > float4Column", 0);
@@ -125,7 +124,7 @@ public class TestComparisonFunctions extends ExecTest {
@Test
public void testFloat8(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
runTest(bitContext, connection, "float8Column == float8Column", 100);
runTest(bitContext, connection, "float8Column != float8Column", 0);
runTest(bitContext, connection, "float8Column > float8Column", 0);
@@ -136,7 +135,7 @@ public class TestComparisonFunctions extends ExecTest {
@Test
public void testIntNullable(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
runTest(bitContext, connection, "intNullableColumn == intNullableColumn", 50);
runTest(bitContext, connection, "intNullableColumn != intNullableColumn", 0);
runTest(bitContext, connection, "intNullableColumn > intNullableColumn", 0);
@@ -144,9 +143,10 @@ public class TestComparisonFunctions extends ExecTest {
runTest(bitContext, connection, "intNullableColumn >= intNullableColumn", 50);
runTest(bitContext, connection, "intNullableColumn <= intNullableColumn", 50);
}
+
@Test
public void testBigIntNullable(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
runTest(bitContext, connection, "bigIntNullableColumn == bigIntNullableColumn", 50);
runTest(bitContext, connection, "bigIntNullableColumn != bigIntNullableColumn", 0);
runTest(bitContext, connection, "bigIntNullableColumn > bigIntNullableColumn", 0);
@@ -154,5 +154,4 @@ public class TestComparisonFunctions extends ExecTest {
runTest(bitContext, connection, "bigIntNullableColumn >= bigIntNullableColumn", 50);
runTest(bitContext, connection, "bigIntNullableColumn <= bigIntNullableColumn", 50);
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestImplicitCastFunctions.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestImplicitCastFunctions.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestImplicitCastFunctions.java
index 03c6f41..7f24c4c 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestImplicitCastFunctions.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestImplicitCastFunctions.java
@@ -26,7 +26,7 @@ import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -43,22 +43,22 @@ import com.google.common.base.Charsets;
import com.google.common.io.Resources;
public class TestImplicitCastFunctions extends ExecTest {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestImplicitCastFunctions.class);
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestImplicitCastFunctions.class);
- DrillConfig c = DrillConfig.create();
- PhysicalPlanReader reader;
- FunctionImplementationRegistry registry;
- FragmentContext context;
+ private final DrillConfig c = DrillConfig.create();
+ private PhysicalPlanReader reader;
+ private FunctionImplementationRegistry registry;
+ private FragmentContext context;
public Object[] getRunResult(SimpleRootExec exec) {
int size = 0;
- for (ValueVector v : exec) {
+ for (final ValueVector v : exec) {
size++;
}
- Object[] res = new Object [size];
+ final Object[] res = new Object [size];
int i = 0;
- for (ValueVector v : exec) {
+ for (final ValueVector v : exec) {
res[i++] = v.getAccessor().getObject(0);
}
return res;
@@ -69,13 +69,13 @@ public class TestImplicitCastFunctions extends ExecTest {
new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- String planString = Resources.toString(Resources.getResource(planPath), Charsets.UTF_8);
+ final String planString = Resources.toString(Resources.getResource(planPath), Charsets.UTF_8);
if (reader == null) {
reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
}
@@ -85,16 +85,16 @@ public class TestImplicitCastFunctions extends ExecTest {
if (context == null) {
context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
}
- PhysicalPlan plan = reader.readPhysicalPlan(planString);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlan plan = reader.readPhysicalPlan(planString);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
exec.next(); // skip schema batch
while (exec.next()) {
- Object [] res = getRunResult(exec);
+ final Object [] res = getRunResult(exec);
assertEquals("return count does not match", res.length, expectedResults.length);
- for (int i = 0; i<res.length; i++) {
+ for (int i = 0; i < res.length; i++) {
assertEquals(String.format("column %s does not match", i), res[i], expectedResults[i]);
}
}
@@ -109,7 +109,7 @@ public class TestImplicitCastFunctions extends ExecTest {
@Test
public void testImplicitCastWithConstant(@Injectable final DrillbitContext bitContext,
@Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[21];
+ final Object [] expected = new Object[21];
expected [0] = new Double (30.1);
expected [1] = new Double (30.1);
expected [2] = new Double (30.1);
@@ -141,7 +141,7 @@ public class TestImplicitCastFunctions extends ExecTest {
@Test
public void testImplicitCastWithMockColumn(@Injectable final DrillbitContext bitContext,
@Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[5];
+ final Object [] expected = new Object[5];
expected [0] = new Integer (0);
expected [1] = new Integer (0);
expected [2] = new Float (-2.14748365E9);
@@ -154,7 +154,7 @@ public class TestImplicitCastFunctions extends ExecTest {
@Test
public void testImplicitCastWithNullExpression(@Injectable final DrillbitContext bitContext,
@Injectable UserServer.UserClientConnection connection) throws Throwable{
- Object [] expected = new Object[10];
+ final Object [] expected = new Object[10];
expected [0] = Boolean.TRUE;
expected [1] = Boolean.FALSE;
@@ -170,5 +170,4 @@ public class TestImplicitCastFunctions extends ExecTest {
runTest(bitContext, connection, expected, "functions/cast/testICastNullExp.json");
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestOptiqPlans.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestOptiqPlans.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestOptiqPlans.java
index dc37071..f2f2f0d 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestOptiqPlans.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestOptiqPlans.java
@@ -30,7 +30,7 @@ import org.apache.drill.exec.client.DrillClient;
import org.apache.drill.exec.coord.ClusterCoordinator;
import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
import org.apache.drill.exec.expr.holders.VarBinaryHolder;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.ops.QueryContext;
import org.apache.drill.exec.opt.BasicOptimizer;
@@ -66,28 +66,28 @@ import com.google.common.io.Resources;
@Ignore
public class TestOptiqPlans extends ExecTest {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestOptiqPlans.class);
- DrillConfig c = DrillConfig.create();
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestOptiqPlans.class);
+ private final DrillConfig config = DrillConfig.create();
@Test
public void orderBy(@Injectable final BootStrapContext ctxt, @Injectable UserClientConnection connection,
@Injectable ClusterCoordinator coord, @Injectable DataConnectionCreator com,
@Injectable Controller controller, @Injectable WorkEventBus workBus) throws Throwable {
- SimpleRootExec exec = doLogicalTest(ctxt, connection, "/logical_order.json", coord, com, controller, workBus);
+ final SimpleRootExec exec = doLogicalTest(ctxt, connection, "/logical_order.json", coord, com, controller, workBus);
}
@Test
public void stringFilter(@Injectable final BootStrapContext ctxt, @Injectable UserClientConnection connection,
@Injectable ClusterCoordinator coord, @Injectable DataConnectionCreator com,
@Injectable Controller controller, @Injectable WorkEventBus workBus) throws Throwable {
- SimpleRootExec exec = doLogicalTest(ctxt, connection, "/logical_string_filter.json", coord, com, controller, workBus);
+ final SimpleRootExec exec = doLogicalTest(ctxt, connection, "/logical_string_filter.json", coord, com, controller, workBus);
}
@Test
public void groupBy(@Injectable final BootStrapContext bitContext, @Injectable UserClientConnection connection,
@Injectable ClusterCoordinator coord, @Injectable DataConnectionCreator com,
@Injectable Controller controller, @Injectable WorkEventBus workBus) throws Throwable {
- SimpleRootExec exec = doLogicalTest(bitContext, connection, "/logical_group.json", coord, com, controller, workBus);
+ final SimpleRootExec exec = doLogicalTest(bitContext, connection, "/logical_group.json", coord, com, controller, workBus);
}
private SimpleRootExec doLogicalTest(final BootStrapContext context, UserClientConnection connection, String file,
@@ -97,24 +97,23 @@ public class TestOptiqPlans extends ExecTest {
context.getMetrics();
result = new MetricRegistry();
context.getAllocator();
- result = new TopLevelAllocator();
+ result = RootAllocatorFactory.newRoot(config);
context.getConfig();
- result = c;
+ result = config;
}
};
- RemoteServiceSet lss = RemoteServiceSet.getLocalServiceSet();
- DrillbitContext bitContext = new DrillbitContext(DrillbitEndpoint.getDefaultInstance(), context, coord, controller,
- com, workBus, new LocalPStoreProvider(DrillConfig.create()), null);
- QueryContext qc = new QueryContext(UserSession.Builder.newBuilder().setSupportComplexTypes(true).build(),
+ final RemoteServiceSet lss = RemoteServiceSet.getLocalServiceSet();
+ final DrillbitContext bitContext = new DrillbitContext(DrillbitEndpoint.getDefaultInstance(), context, coord, controller,
+ com, workBus, new LocalPStoreProvider(config), null);
+ final QueryContext qc = new QueryContext(UserSession.Builder.newBuilder().setSupportComplexTypes(true).build(),
bitContext);
- PhysicalPlanReader reader = bitContext.getPlanReader();
- LogicalPlan plan = reader.readLogicalPlan(Files.toString(FileUtils.getResourceAsFile(file), Charsets.UTF_8));
- PhysicalPlan pp = new BasicOptimizer(qc, connection).optimize(new BasicOptimizer.BasicOptimizationContext(qc), plan);
+ final PhysicalPlanReader reader = bitContext.getPlanReader();
+ final LogicalPlan plan = reader.readLogicalPlan(Files.toString(FileUtils.getResourceAsFile(file), Charsets.UTF_8));
+ final PhysicalPlan pp = new BasicOptimizer(qc, connection).optimize(new BasicOptimizer.BasicOptimizationContext(qc), plan);
-
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext fctxt = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(fctxt, (FragmentRoot) pp.getSortedOperators(false)
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(config);
+ final FragmentContext fctxt = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(fctxt, (FragmentRoot) pp.getSortedOperators(false)
.iterator().next()));
return exec;
@@ -122,24 +121,23 @@ public class TestOptiqPlans extends ExecTest {
@Test
public void testFilterPlan() throws Exception {
- RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- DrillConfig config = DrillConfig.create();
+ final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- try (Drillbit bit1 = new Drillbit(config, serviceSet);
- DrillClient client = new DrillClient(config, serviceSet.getCoordinator());) {
+ try (final Drillbit bit1 = new Drillbit(config, serviceSet);
+ final DrillClient client = new DrillClient(config, serviceSet.getCoordinator());) {
bit1.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
Resources.toString(Resources.getResource("physical_filter.json"), Charsets.UTF_8));
- RecordBatchLoader loader = new RecordBatchLoader(bit1.getContext().getAllocator());
- for (QueryDataBatch b : results) {
+ final RecordBatchLoader loader = new RecordBatchLoader(bit1.getContext().getAllocator());
+ for (final QueryDataBatch b : results) {
System.out.println(String.format("Got %d results", b.getHeader().getRowCount()));
loader.load(b.getHeader().getDef(), b.getData());
- for (VectorWrapper<?> vw : loader) {
+ for (final VectorWrapper<?> vw : loader) {
System.out.println(vw.getValueVector().getField().toExpr());
- ValueVector vv = vw.getValueVector();
+ final ValueVector vv = vw.getValueVector();
for (int i = 0; i < vv.getAccessor().getValueCount(); i++) {
- Object o = vv.getAccessor().getObject(i);
+ final Object o = vv.getAccessor().getObject(i);
System.out.println(vv.getAccessor().getObject(i));
}
}
@@ -152,24 +150,23 @@ public class TestOptiqPlans extends ExecTest {
@Test
public void testJoinPlan() throws Exception {
- RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- DrillConfig config = DrillConfig.create();
+ final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- try (Drillbit bit1 = new Drillbit(config, serviceSet);
- DrillClient client = new DrillClient(config, serviceSet.getCoordinator());) {
+ try (final Drillbit bit1 = new Drillbit(config, serviceSet);
+ final DrillClient client = new DrillClient(config, serviceSet.getCoordinator());) {
bit1.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
Resources.toString(Resources.getResource("physical_join.json"), Charsets.UTF_8));
- RecordBatchLoader loader = new RecordBatchLoader(bit1.getContext().getAllocator());
- for (QueryDataBatch b : results) {
+ final RecordBatchLoader loader = new RecordBatchLoader(bit1.getContext().getAllocator());
+ for (final QueryDataBatch b : results) {
System.out.println(String.format("Got %d results", b.getHeader().getRowCount()));
loader.load(b.getHeader().getDef(), b.getData());
- for (VectorWrapper<?> vw : loader) {
+ for (final VectorWrapper<?> vw : loader) {
System.out.println(vw.getValueVector().getField().toExpr());
- ValueVector vv = vw.getValueVector();
+ final ValueVector vv = vw.getValueVector();
for (int i = 0; i < vv.getAccessor().getValueCount(); i++) {
- Object o = vv.getAccessor().getObject(i);
+ final Object o = vv.getAccessor().getObject(i);
System.out.println(vv.getAccessor().getObject(i));
}
}
@@ -182,27 +179,26 @@ public class TestOptiqPlans extends ExecTest {
@Test
public void testFilterString() throws Exception {
- RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- DrillConfig config = DrillConfig.create();
+ final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- try (Drillbit bit1 = new Drillbit(config, serviceSet);
- DrillClient client = new DrillClient(config, serviceSet.getCoordinator());) {
+ try (final Drillbit bit1 = new Drillbit(config, serviceSet);
+ final DrillClient client = new DrillClient(config, serviceSet.getCoordinator());) {
bit1.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.LOGICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.LOGICAL,
Resources.toString(Resources.getResource("logical_string_filter.json"), Charsets.UTF_8));
- RecordBatchLoader loader = new RecordBatchLoader(bit1.getContext().getAllocator());
- for (QueryDataBatch b : results) {
+ final RecordBatchLoader loader = new RecordBatchLoader(bit1.getContext().getAllocator());
+ for (final QueryDataBatch b : results) {
System.out.println(String.format("Got %d results", b.getHeader().getRowCount()));
loader.load(b.getHeader().getDef(), b.getData());
- for (VectorWrapper<?> vw : loader) {
+ for (final VectorWrapper<?> vw : loader) {
System.out.println(vw.getValueVector().getField().toExpr());
- ValueVector vv = vw.getValueVector();
+ final ValueVector vv = vw.getValueVector();
for (int i = 0; i < vv.getAccessor().getValueCount(); i++) {
- Object o = vv.getAccessor().getObject(i);
+ final Object o = vv.getAccessor().getObject(i);
if (vv instanceof VarBinaryVector) {
- VarBinaryVector.Accessor x = ((VarBinaryVector) vv).getAccessor();
- VarBinaryHolder vbh = new VarBinaryHolder();
+ final VarBinaryVector.Accessor x = ((VarBinaryVector) vv).getAccessor();
+ final VarBinaryHolder vbh = new VarBinaryHolder();
x.get(i, vbh);
System.out.printf("%d..%d", vbh.start, vbh.end);
@@ -222,27 +218,26 @@ public class TestOptiqPlans extends ExecTest {
@Test
public void testLogicalJsonScan() throws Exception {
- RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- DrillConfig config = DrillConfig.create();
+ final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- try (Drillbit bit1 = new Drillbit(config, serviceSet);
- DrillClient client = new DrillClient(config, serviceSet.getCoordinator());) {
+ try (final Drillbit bit1 = new Drillbit(config, serviceSet);
+ final DrillClient client = new DrillClient(config, serviceSet.getCoordinator());) {
bit1.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.LOGICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.LOGICAL,
Resources.toString(Resources.getResource("logical_json_scan.json"), Charsets.UTF_8));
- RecordBatchLoader loader = new RecordBatchLoader(bit1.getContext().getAllocator());
- for (QueryDataBatch b : results) {
+ final RecordBatchLoader loader = new RecordBatchLoader(bit1.getContext().getAllocator());
+ for (final QueryDataBatch b : results) {
System.out.println(String.format("Got %d results", b.getHeader().getRowCount()));
loader.load(b.getHeader().getDef(), b.getData());
- for (VectorWrapper vw : loader) {
+ for (final VectorWrapper vw : loader) {
System.out.println(vw.getValueVector().getField().toExpr());
- ValueVector vv = vw.getValueVector();
+ final ValueVector vv = vw.getValueVector();
for (int i = 0; i < vv.getAccessor().getValueCount(); i++) {
- Object o = vv.getAccessor().getObject(i);
+ final Object o = vv.getAccessor().getObject(i);
if (vv instanceof VarBinaryVector) {
- VarBinaryVector.Accessor x = ((VarBinaryVector) vv).getAccessor();
- VarBinaryHolder vbh = new VarBinaryHolder();
+ final VarBinaryVector.Accessor x = ((VarBinaryVector) vv).getAccessor();
+ final VarBinaryHolder vbh = new VarBinaryHolder();
x.get(i, vbh);
System.out.printf("%d..%d", vbh.start, vbh.end);
@@ -262,27 +257,26 @@ public class TestOptiqPlans extends ExecTest {
@Test
public void testOrderVarbinary() throws Exception {
- RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- DrillConfig config = DrillConfig.create();
+ final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- try (Drillbit bit1 = new Drillbit(config, serviceSet);
- DrillClient client = new DrillClient(config, serviceSet.getCoordinator());) {
+ try (final Drillbit bit1 = new Drillbit(config, serviceSet);
+ final DrillClient client = new DrillClient(config, serviceSet.getCoordinator());) {
bit1.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+ final List<QueryDataBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
Resources.toString(Resources.getResource("physical_order_varbinary.json"), Charsets.UTF_8));
- RecordBatchLoader loader = new RecordBatchLoader(bit1.getContext().getAllocator());
- for (QueryDataBatch b : results) {
+ final RecordBatchLoader loader = new RecordBatchLoader(bit1.getContext().getAllocator());
+ for (final QueryDataBatch b : results) {
System.out.println(String.format("Got %d results", b.getHeader().getRowCount()));
loader.load(b.getHeader().getDef(), b.getData());
- for (VectorWrapper vw : loader) {
+ for (final VectorWrapper vw : loader) {
System.out.println(vw.getValueVector().getField().toExpr());
- ValueVector vv = vw.getValueVector();
+ final ValueVector vv = vw.getValueVector();
for (int i = 0; i < vv.getAccessor().getValueCount(); i++) {
- Object o = vv.getAccessor().getObject(i);
+ final Object o = vv.getAccessor().getObject(i);
if (vv instanceof VarBinaryVector) {
- VarBinaryVector.Accessor x = ((VarBinaryVector) vv).getAccessor();
- VarBinaryHolder vbh = new VarBinaryHolder();
+ final VarBinaryVector.Accessor x = ((VarBinaryVector) vv).getAccessor();
+ final VarBinaryHolder vbh = new VarBinaryHolder();
x.get(i, vbh);
System.out.printf("%d..%d", vbh.start, vbh.end);
@@ -307,22 +301,21 @@ public class TestOptiqPlans extends ExecTest {
bitContext.getMetrics();
result = new MetricRegistry();
bitContext.getAllocator();
- result = new TopLevelAllocator();
+ result = RootAllocatorFactory.newRoot(config);
bitContext.getConfig();
- result = c;
+ result = config;
}
};
- StoragePluginRegistry reg = new StoragePluginRegistry(bitContext);
+ final StoragePluginRegistry reg = new StoragePluginRegistry(bitContext);
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(),
+ final PhysicalPlanReader reader = new PhysicalPlanReader(config, config.getMapper(),
CoordinationProtos.DrillbitEndpoint.getDefaultInstance(), reg);
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile(file), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false)
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile(file), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(config);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false)
.iterator().next()));
return exec;
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFragmentRun.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFragmentRun.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFragmentRun.java
index 09ba1a5..5d0ea2d 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFragmentRun.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFragmentRun.java
@@ -21,12 +21,11 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
-import java.nio.charset.Charset;
import java.util.List;
import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.client.DrillClient;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.pop.PopUnitTestBase;
import org.apache.drill.exec.proto.UserBitShared.QueryType;
import org.apache.drill.exec.record.RecordBatchLoader;
@@ -41,34 +40,32 @@ import com.google.common.base.Charsets;
import com.google.common.io.Files;
public class TestSimpleFragmentRun extends PopUnitTestBase {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestSimpleFragmentRun.class);
-
- private static final Charset UTF_8 = Charset.forName("UTF-8");
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestSimpleFragmentRun.class);
@Test
public void runNoExchangeFragment() throws Exception {
- try (RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- Drillbit bit = new Drillbit(CONFIG, serviceSet);
- DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
+ try (final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+ final Drillbit bit = new Drillbit(CONFIG, serviceSet);
+ final DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
// run query.
bit.run();
client.connect();
- String path = "/physical_test2.json";
+ final String path = "/physical_test2.json";
// String path = "/filter/test1.json";
- List<QueryDataBatch> results = client.runQuery(QueryType.PHYSICAL, Files.toString(FileUtils.getResourceAsFile(path), Charsets.UTF_8));
+ final List<QueryDataBatch> results = client.runQuery(QueryType.PHYSICAL, Files.toString(FileUtils.getResourceAsFile(path), Charsets.UTF_8));
// look at records
- RecordBatchLoader batchLoader = new RecordBatchLoader(client.getAllocator());
+ final RecordBatchLoader batchLoader = new RecordBatchLoader(client.getAllocator());
int recordCount = 0;
- for (QueryDataBatch batch : results) {
- boolean schemaChanged = batchLoader.load(batch.getHeader().getDef(), batch.getData());
+ for (final QueryDataBatch batch : results) {
+ final boolean schemaChanged = batchLoader.load(batch.getHeader().getDef(), batch.getData());
boolean firstColumn = true;
// print headers.
if (schemaChanged) {
System.out.println("\n\n========NEW SCHEMA=========\n\n");
- for (VectorWrapper<?> value : batchLoader) {
+ for (final VectorWrapper<?> value : batchLoader) {
if (firstColumn) {
firstColumn = false;
@@ -86,7 +83,7 @@ public class TestSimpleFragmentRun extends PopUnitTestBase {
for (int i = 0; i < batchLoader.getRecordCount(); i++) {
boolean first = true;
recordCount++;
- for (VectorWrapper<?> value : batchLoader) {
+ for (final VectorWrapper<?> value : batchLoader) {
if (first) {
first = false;
} else {
@@ -108,20 +105,20 @@ public class TestSimpleFragmentRun extends PopUnitTestBase {
@Test
public void runJSONScanPopFragment() throws Exception {
- try (RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
- Drillbit bit = new Drillbit(CONFIG, serviceSet);
- DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator())) {
+ try (final RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+ final Drillbit bit = new Drillbit(CONFIG, serviceSet);
+ final DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator())) {
// run query.
bit.run();
client.connect();
- List<QueryDataBatch> results = client.runQuery(QueryType.PHYSICAL,
+ final List<QueryDataBatch> results = client.runQuery(QueryType.PHYSICAL,
Files.toString(FileUtils.getResourceAsFile("/physical_json_scan_test1.json"), Charsets.UTF_8)
.replace("#{TEST_FILE}", FileUtils.getResourceAsFile("/scan_json_test_1.json").toURI().toString())
);
// look at records
- RecordBatchLoader batchLoader = new RecordBatchLoader(new TopLevelAllocator(CONFIG));
+ final RecordBatchLoader batchLoader = new RecordBatchLoader(RootAllocatorFactory.newRoot(CONFIG));
int recordCount = 0;
//int expectedBatchCount = 2;
@@ -129,7 +126,7 @@ public class TestSimpleFragmentRun extends PopUnitTestBase {
//assertEquals(expectedBatchCount, results.size());
for (int i = 0; i < results.size(); ++i) {
- QueryDataBatch batch = results.get(i);
+ final QueryDataBatch batch = results.get(i);
if (i == 0) {
assertTrue(batch.hasData());
} else {
@@ -143,7 +140,7 @@ public class TestSimpleFragmentRun extends PopUnitTestBase {
// print headers.
System.out.println("\n\n========NEW SCHEMA=========\n\n");
- for (VectorWrapper<?> v : batchLoader) {
+ for (final VectorWrapper<?> v : batchLoader) {
if (firstColumn) {
firstColumn = false;
@@ -162,14 +159,14 @@ public class TestSimpleFragmentRun extends PopUnitTestBase {
for (int r = 0; r < batchLoader.getRecordCount(); r++) {
boolean first = true;
recordCount++;
- for (VectorWrapper<?> v : batchLoader) {
+ for (final VectorWrapper<?> v : batchLoader) {
if (first) {
first = false;
} else {
System.out.print("\t");
}
- ValueVector.Accessor accessor = v.getValueVector().getAccessor();
+ final ValueVector.Accessor accessor = v.getValueVector().getAccessor();
System.out.print(accessor.getObject(r));
}
if (!first) {
@@ -183,5 +180,4 @@ public class TestSimpleFragmentRun extends PopUnitTestBase {
assertEquals(2, recordCount);
}
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bfe6cfad/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFunctions.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFunctions.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFunctions.java
index d551319..4cbd4a9 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFunctions.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFunctions.java
@@ -44,7 +44,7 @@ import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
import org.apache.drill.exec.expr.fn.impl.StringFunctionHelpers;
import org.apache.drill.exec.expr.holders.NullableVarBinaryHolder;
import org.apache.drill.exec.expr.holders.NullableVarCharHolder;
-import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.FragmentRoot;
@@ -65,13 +65,12 @@ import com.google.common.io.Files;
import com.sun.codemodel.JClassAlreadyExistsException;
public class TestSimpleFunctions extends ExecTest {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestSimpleFunctions.class);
-
- DrillConfig c = DrillConfig.create();
+ //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestSimpleFunctions.class);
+ private final DrillConfig c = DrillConfig.create();
@Test
public void testHashFunctionResolution(@Injectable DrillConfig config) throws JClassAlreadyExistsException, IOException {
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(config);
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(config);
// test required vs nullable Int input
resolveHash(config,
new TypedNullConstant(Types.optional(TypeProtos.MinorType.INT)),
@@ -136,46 +135,44 @@ public class TestSimpleFunctions extends ExecTest {
public void resolveHash(DrillConfig config, LogicalExpression arg, TypeProtos.MajorType expectedArg,
TypeProtos.MajorType expectedOut, TypeProtos.DataMode expectedBestInputMode,
FunctionImplementationRegistry registry) throws JClassAlreadyExistsException, IOException {
- List<LogicalExpression> args = new ArrayList<>();
+ final List<LogicalExpression> args = new ArrayList<>();
args.add(arg);
- String[] registeredNames = { "hash" };
+ final String[] registeredNames = { "hash" };
FunctionCall call = new FunctionCall(
"hash",
args,
ExpressionPosition.UNKNOWN
);
- FunctionResolver resolver = FunctionResolverFactory.getResolver(call);
- DrillFuncHolder matchedFuncHolder = registry.findDrillFunction(resolver, call);
+ final FunctionResolver resolver = FunctionResolverFactory.getResolver(call);
+ final DrillFuncHolder matchedFuncHolder = registry.findDrillFunction(resolver, call);
assertEquals( expectedBestInputMode, matchedFuncHolder.getParmMajorType(0).getMode());
}
@Test
public void testSubstring(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
-
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
new NonStrictExpectations(){{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/testSubstring.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/testSubstring.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- while(exec.next()){
- NullableVarCharVector c1 = exec.getValueVectorById(new SchemaPath("col3", ExpressionPosition.UNKNOWN), NullableVarCharVector.class);
- NullableVarCharVector.Accessor a1;
- a1 = c1.getAccessor();
+ while(exec.next()) {
+ final NullableVarCharVector c1 = exec.getValueVectorById(new SchemaPath("col3", ExpressionPosition.UNKNOWN), NullableVarCharVector.class);
+ final NullableVarCharVector.Accessor a1 = c1.getAccessor();
int count = 0;
- for(int i = 0; i < c1.getAccessor().getValueCount(); i++){
+ for(int i = 0; i < c1.getAccessor().getValueCount(); i++) {
if (!a1.isNull(i)) {
- NullableVarCharHolder holder = new NullableVarCharHolder();
+ final NullableVarCharHolder holder = new NullableVarCharHolder();
a1.get(i, holder);
assertEquals("aaaa", StringFunctionHelpers.toStringFromUTF8(holder.start, holder.end, holder.buffer));
++count;
@@ -184,40 +181,37 @@ public class TestSimpleFunctions extends ExecTest {
assertEquals(50, count);
}
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
-
}
@Test
public void testSubstringNegative(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
-
- new NonStrictExpectations(){{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/testSubstringNegative.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/testSubstringNegative.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- while(exec.next()){
- NullableVarCharVector c1 = exec.getValueVectorById(new SchemaPath("col3", ExpressionPosition.UNKNOWN), NullableVarCharVector.class);
- NullableVarCharVector.Accessor a1;
- a1 = c1.getAccessor();
+ while(exec.next()) {
+ final NullableVarCharVector c1 = exec.getValueVectorById(new SchemaPath("col3", ExpressionPosition.UNKNOWN), NullableVarCharVector.class);
+ final NullableVarCharVector.Accessor a1 = c1.getAccessor();
int count = 0;
- for(int i = 0; i < c1.getAccessor().getValueCount(); i++){
+ for(int i = 0; i < c1.getAccessor().getValueCount(); i++) {
if (!a1.isNull(i)) {
- NullableVarCharHolder holder = new NullableVarCharHolder();
+ final NullableVarCharHolder holder = new NullableVarCharHolder();
a1.get(i, holder);
//when offset is negative, substring return empty string.
assertEquals("", StringFunctionHelpers.toStringFromUTF8(holder.start, holder.end, holder.buffer));
@@ -227,40 +221,37 @@ public class TestSimpleFunctions extends ExecTest {
assertEquals(50, count);
}
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
-
}
@Test
public void testByteSubstring(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
-
- new NonStrictExpectations(){{
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+ new NonStrictExpectations() {{
bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getAllocator(); result = RootAllocatorFactory.newRoot(c);
bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
bitContext.getConfig(); result = c;
bitContext.getCompiler(); result = CodeCompiler.getTestCompiler(c);
}};
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/testByteSubstring.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ final PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ final PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/testByteSubstring.json"), Charsets.UTF_8));
+ final FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ final FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ final SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
- while(exec.next()){
- NullableVarBinaryVector c1 = exec.getValueVectorById(new SchemaPath("col3", ExpressionPosition.UNKNOWN), NullableVarBinaryVector.class);
- NullableVarBinaryVector.Accessor a1;
- a1 = c1.getAccessor();
+ while(exec.next()) {
+ final NullableVarBinaryVector c1 = exec.getValueVectorById(new SchemaPath("col3", ExpressionPosition.UNKNOWN), NullableVarBinaryVector.class);
+ final NullableVarBinaryVector.Accessor a1 = c1.getAccessor();
int count = 0;
- for(int i = 0; i < c1.getAccessor().getValueCount(); i++){
+ for(int i = 0; i < c1.getAccessor().getValueCount(); i++) {
if (!a1.isNull(i)) {
- NullableVarBinaryHolder holder = new NullableVarBinaryHolder();
+ final NullableVarBinaryHolder holder = new NullableVarBinaryHolder();
a1.get(i, holder);
assertEquals("aa", StringFunctionHelpers.toStringFromUTF8(holder.start, holder.end, holder.buffer));
++count;
@@ -269,11 +260,9 @@ public class TestSimpleFunctions extends ExecTest {
assertEquals(50, count);
}
- if(context.getFailureCause() != null){
+ if(context.getFailureCause() != null) {
throw context.getFailureCause();
}
assertTrue(!context.isFailed());
-
}
-
}