[GitHub] drill pull request #1110: DRILL-6115: SingleMergeExchange is not scaling up ...

[HanumathRao]

GitHub user HanumathRao opened a pull request:

https://github.com/apache/drill/pull/1110

DRILL-6115: SingleMergeExchange is not scaling up when many minor fra…

…gments are allocated for a query.

Currently a singlemerge exchange is merging all the fragment streams on 
foreman. This can cause cpu bottleneck and also huge memory consumption at the 
foreman. 

This PR contains changes to introduce a new Multiplex Operator called 
OrderedMuxExchange which merges the minor fragment streams pertaining to one 
drillbit and send as one output stream to the foreman. 

The existing multiplex mechanism is used to introduce these operators.

Please review this PR.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/HanumathRao/drill DRILL-6115

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/drill/pull/1110.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #1110


commit 43a71277aeec9bb377181728b2ce563437d7e46d
Author: hmaduri 
Date:   2018-01-22T00:42:28Z

DRILL-6115: SingleMergeExchange is not scaling up when many minor fragments 
are allocated for a query.




---

Re: Error in Drill 1.13

[Charles Givre]

Thanks Tim! I got it working!

Sent from my iPhone

> On Feb 2, 2018, at 15:11, Timothy Farkas  wrote:
> 
> Hi Charles,
> 
> FragmentContext used to be a concrete class. Now the FragmentContext has been 
> changed to an interface, and the concrete class that implements it is now 
> FragmentContextImpl
> 
> Thanks,
> Tim
> 
> 
> From: Charles Givre 
> Sent: Friday, February 2, 2018 11:47:12 AM
> To: dev@drill.apache.org
> Subject: Error in Drill 1.13
> 
> Hello all,
> I’m getting ready to submit a PR for a log format plugin for Drill and after 
> I rebased Drill, I’m now getting the following error:
> 
> 
> java.lang.IncompatibleClassChangeError: Found interface 
> org.apache.drill.exec.ops.FragmentContext, but class was expected
>at 
> org.apache.drill.exec.store.log.LogRecordReader.(LogRecordReader.java:90)
>at 
> org.apache.drill.exec.store.log.LogFormatPlugin.getRecordReader(LogFormatPlugin.java:63)
>at 
> org.apache.drill.exec.store.dfs.easy.EasyFormatPlugin.getReaderBatch(EasyFormatPlugin.java:150)
>at 
> org.apache.drill.exec.store.dfs.easy.EasyReaderBatchCreator.getBatch(EasyReaderBatchCreator.java:33)
>at 
> org.apache.drill.exec.store.dfs.easy.EasyReaderBatchCreator.getBatch(EasyReaderBatchCreator.java:28)
>at 
> org.apache.drill.exec.physical.impl.ImplCreator.getRecordBatch(ImplCreator.java:159)
>at 
> org.apache.drill.exec.physical.impl.ImplCreator.getChildren(ImplCreator.java:182)
>at 
> org.apache.drill.exec.physical.impl.ImplCreator.getRecordBatch(ImplCreator.java:137)
>at 
> org.apache.drill.exec.physical.impl.ImplCreator.getChildren(ImplCreator.java:182)
>at 
> org.apache.drill.exec.physical.impl.ImplCreator.getRecordBatch(ImplCreator.java:137)
>at 
> org.apache.drill.exec.physical.impl.ImplCreator.getChildren(ImplCreator.java:182)
>at 
> org.apache.drill.exec.physical.impl.ImplCreator.getRootExec(ImplCreator.java:110)
>at 
> org.apache.drill.exec.physical.impl.ImplCreator.getExec(ImplCreator.java:87)
>at 
> org.apache.drill.exec.work.fragment.FragmentExecutor.run(FragmentExecutor.java:206)
>at 
> org.apache.drill.common.SelfCleaningRunnable.run(SelfCleaningRunnable.java:38)
>at 
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
>at 
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
>at java.lang.Thread.run(Thread.java:745)
> 
> 
> Here is the code which seems to be causing the problem:
> 
> public LogRecordReader(FragmentContext fragmentContext, String inputPath, 
> DrillFileSystem fileSystem,
>   List columns, 
> LogFormatPlugin.LogFormatConfig config) throws OutOfMemoryException {
>  try {
>Path hdfsPath = new Path(inputPath);
>Configuration conf = new Configuration();
>FSDataInputStream fsStream = fileSystem.open(hdfsPath);
>CompressionCodecFactory factory = new CompressionCodecFactory(conf);
>CompressionCodec codec = factory.getCodec(hdfsPath);
>if (codec == null) {
>  reader = new BufferedReader(new 
> InputStreamReader(fsStream.getWrappedStream(), "UTF-8"));
>} else {
>  CompressionInputStream comInputStream = 
> codec.createInputStream(fsStream.getWrappedStream());
>  reader = new BufferedReader(new InputStreamReader(comInputStream));
>}
>this.inputPath = inputPath;
>this.lineCount = 0;
>this.config = config;
>this.buffer = fragmentContext.getManagedBuffer(4096);
>setColumns(columns);
> 
>  } catch (IOException e) {
>logger.debug("Log Reader Plugin: " + e.getMessage());
>  }
> }
> and here is a link to the repo with the complete code; 
> https://urldefense.proofpoint.com/v2/url?u=https-3A__github.com_cgivre_drill-2Dlogfile-2Dplugin=DwIFaQ=cskdkSMqhcnjZxdQVpwTXg=4eQVr8zB8ZBff-yxTimdOQ=Lz6tyq_4Ljazq01AWJA1ZTV4s9ysO1lcLbVMA3M9jAs=Qx0c4Si8Vr7fI-2tNtB6waqrqIiLnvLDaFOoRJr-354=
>  
> .
>   I’m a little stumped on this and would appreciate any suggestions.  My 
> plugin DOES work on 1.12.
> —C
> 
>