[GitHub] incubator-carbondata pull request #333: Optimized no kettle flow and fixed i...

[qiuchenjian-2]

GitHub user ravipesala opened a pull request:

    https://github.com/apache/incubator-carbondata/pull/333

    Optimized no kettle flow and fixed issues in cluster

    Optimized no kettle flow and fixed issues in cluster

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

    $ git pull https://github.com/ravipesala/incubator-carbondata optimize-no-kettle

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

    https://github.com/apache/incubator-carbondata/pull/333.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 #333
   
----
commit d7e13993401e4120adbeaef7b8923fc031123aea
Author: ravipesala <[hidden email]>
Date:   2016-11-19T04:24:56Z

    Optimized no kettle flow and fixed issues in cluster

----


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user ravipesala commented on the issue:

    https://github.com/apache/incubator-carbondata/pull/333
 
    http://136.243.101.176:8080/job/ApacheCarbonManualPRBuilder/630/


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user jackylk commented on the issue:

    https://github.com/apache/incubator-carbondata/pull/333
 
    Can you add more description, what is the optimization?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user chenliang613 commented on the issue:

    https://github.com/apache/incubator-carbondata/pull/333
 
    @ravipesala  and please use the correct format for PR title


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user ravipesala commented on the issue:

    https://github.com/apache/incubator-carbondata/pull/333
 
    Updated PR, please review it.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/incubator-carbondata/pull/333#discussion_r90587816
 
    --- Diff: hadoop/src/main/java/org/apache/carbondata/hadoop/csv/CSVInputFormat.java ---
    @@ -138,6 +140,17 @@ public static void setQuoteCharacter(String quoteCharacter, Configuration config
       }
     
       /**
    +   * Sets the read buffer size to configuration.
    +   * @param bufferSize
    +   * @param configuration
    +   */
    +  public static void setReadBufferSize(String bufferSize, Configuration configuration) {
    --- End diff --
   
    why bufferSize is string but not int?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/incubator-carbondata/pull/333#discussion_r90588002
 
    --- Diff: processing/src/main/java/org/apache/carbondata/processing/newflow/iterator/InputIterator.java ---
    @@ -0,0 +1,40 @@
    +/*
    + * 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.carbondata.processing.newflow.iterator;
    +
    +import java.util.Iterator;
    +
    +/**
    + * It iterates the data of record readers
    + */
    +public abstract class InputIterator<E> implements Iterator<E> {
    --- End diff --
   
    Can we add initialize and close function to `CarbonIterator` so that that can be reused?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/incubator-carbondata/pull/333#discussion_r90588312
 
    --- Diff: processing/src/main/java/org/apache/carbondata/processing/newflow/steps/InputProcessorStepImpl.java ---
    @@ -80,40 +87,50 @@ public void initialize() throws CarbonDataLoadingException {
         return iterators;
       }
     
    -  @Override
    -  protected CarbonRow processRow(CarbonRow row) {
    +  @Override protected CarbonRow processRow(CarbonRow row) {
         return null;
       }
     
    +  @Override public void close() {
    +    executorService.shutdown();
    +  }
    +
       /**
        * This iterator wraps the list of iterators and it starts iterating the each
        * iterator of the list one by one. It also parse the data while iterating it.
        */
       private static class InputProcessorIterator extends CarbonIterator<CarbonRowBatch> {
     
    -    private List<Iterator<Object[]>> inputIterators;
    +    private List<InputIterator<Object[]>> inputIterators;
     
    -    private Iterator<Object[]> currentIterator;
    +    private InputIterator<Object[]> currentIterator;
     
         private int counter;
     
         private int batchSize;
     
         private RowParser rowParser;
     
    -    public InputProcessorIterator(List<Iterator<Object[]>> inputIterators,
    -        RowParser rowParser, int batchSize) {
    +    private Future<CarbonRowBatch> future;
    +
    +    private ExecutorService executorService;
    +
    +    private boolean nextBatch = false;
    --- End diff --
   
    initialize in constructor, like counter


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/incubator-carbondata/pull/333#discussion_r90588344
 
    --- Diff: processing/src/main/java/org/apache/carbondata/processing/newflow/steps/InputProcessorStepImpl.java ---
    @@ -80,40 +87,50 @@ public void initialize() throws CarbonDataLoadingException {
         return iterators;
       }
     
    -  @Override
    -  protected CarbonRow processRow(CarbonRow row) {
    +  @Override protected CarbonRow processRow(CarbonRow row) {
         return null;
       }
     
    +  @Override public void close() {
    +    executorService.shutdown();
    +  }
    +
       /**
        * This iterator wraps the list of iterators and it starts iterating the each
        * iterator of the list one by one. It also parse the data while iterating it.
        */
       private static class InputProcessorIterator extends CarbonIterator<CarbonRowBatch> {
     
    -    private List<Iterator<Object[]>> inputIterators;
    +    private List<InputIterator<Object[]>> inputIterators;
     
    -    private Iterator<Object[]> currentIterator;
    +    private InputIterator<Object[]> currentIterator;
     
         private int counter;
     
         private int batchSize;
     
         private RowParser rowParser;
     
    -    public InputProcessorIterator(List<Iterator<Object[]>> inputIterators,
    -        RowParser rowParser, int batchSize) {
    +    private Future<CarbonRowBatch> future;
    +
    +    private ExecutorService executorService;
    +
    +    private boolean nextBatch = false;
    +
    +    public InputProcessorIterator(List<InputIterator<Object[]>> inputIterators,
    +        RowParser rowParser, int batchSize, ExecutorService executorService) {
           this.inputIterators = inputIterators;
           this.batchSize = batchSize;
           this.rowParser = rowParser;
           this.counter = 0;
           // Get the first iterator from the list.
           currentIterator = inputIterators.get(counter++);
    +      currentIterator.initialize();
    +      this.executorService = executorService;
         }
     
    -    @Override
    -    public boolean hasNext() {
    -      return internalHasNext();
    +    @Override public boolean hasNext() {
    --- End diff --
   
    put override to previous line


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/incubator-carbondata/pull/333#discussion_r90588348
 
    --- Diff: processing/src/main/java/org/apache/carbondata/processing/newflow/steps/InputProcessorStepImpl.java ---
    @@ -122,24 +139,52 @@ private boolean internalHasNext() {
           if (!hasNext) {
             // Check next iterator is available in the list.
             if (counter < inputIterators.size()) {
    +          // close the old iterator
    +          currentIterator.close();
               // Get the next iterator from the list.
               currentIterator = inputIterators.get(counter++);
    +          // Initialize the new iterator
    +          currentIterator.initialize();
               hasNext = internalHasNext();
             }
           }
           return hasNext;
         }
     
    -    @Override
    -    public CarbonRowBatch next() {
    -      // Create batch and fill it.
    -      CarbonRowBatch carbonRowBatch = new CarbonRowBatch();
    -      int count = 0;
    -      while (internalHasNext() && count < batchSize) {
    -        carbonRowBatch.addRow(new CarbonRow(rowParser.parseRow(currentIterator.next())));
    -        count++;
    +    @Override public CarbonRowBatch next() {
    +      CarbonRowBatch result = null;
    +      try {
    --- End diff --
   
    put override to previous line


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/incubator-carbondata/pull/333#discussion_r90588408
 
    --- Diff: processing/src/main/java/org/apache/carbondata/processing/newflow/steps/InputProcessorStepImpl.java ---
    @@ -122,24 +139,52 @@ private boolean internalHasNext() {
           if (!hasNext) {
             // Check next iterator is available in the list.
             if (counter < inputIterators.size()) {
    +          // close the old iterator
    +          currentIterator.close();
               // Get the next iterator from the list.
               currentIterator = inputIterators.get(counter++);
    +          // Initialize the new iterator
    +          currentIterator.initialize();
               hasNext = internalHasNext();
             }
           }
           return hasNext;
         }
     
    -    @Override
    -    public CarbonRowBatch next() {
    -      // Create batch and fill it.
    -      CarbonRowBatch carbonRowBatch = new CarbonRowBatch();
    -      int count = 0;
    -      while (internalHasNext() && count < batchSize) {
    -        carbonRowBatch.addRow(new CarbonRow(rowParser.parseRow(currentIterator.next())));
    -        count++;
    +    @Override public CarbonRowBatch next() {
    --- End diff --
   
    put override to previous line


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/incubator-carbondata/pull/333#discussion_r90588425
 
    --- Diff: processing/src/main/java/org/apache/carbondata/processing/newflow/steps/InputProcessorStepImpl.java ---
    @@ -122,24 +139,52 @@ private boolean internalHasNext() {
           if (!hasNext) {
             // Check next iterator is available in the list.
             if (counter < inputIterators.size()) {
    +          // close the old iterator
    +          currentIterator.close();
               // Get the next iterator from the list.
               currentIterator = inputIterators.get(counter++);
    +          // Initialize the new iterator
    +          currentIterator.initialize();
               hasNext = internalHasNext();
             }
           }
           return hasNext;
         }
     
    -    @Override
    -    public CarbonRowBatch next() {
    -      // Create batch and fill it.
    -      CarbonRowBatch carbonRowBatch = new CarbonRowBatch();
    -      int count = 0;
    -      while (internalHasNext() && count < batchSize) {
    -        carbonRowBatch.addRow(new CarbonRow(rowParser.parseRow(currentIterator.next())));
    -        count++;
    +    @Override public CarbonRowBatch next() {
    +      CarbonRowBatch result = null;
    +      try {
    --- End diff --
   
    limit the try scope to `future.get` only


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/incubator-carbondata/pull/333#discussion_r90588455
 
    --- Diff: processing/src/main/java/org/apache/carbondata/processing/newflow/steps/InputProcessorStepImpl.java ---
    @@ -122,24 +139,52 @@ private boolean internalHasNext() {
           if (!hasNext) {
             // Check next iterator is available in the list.
             if (counter < inputIterators.size()) {
    +          // close the old iterator
    +          currentIterator.close();
               // Get the next iterator from the list.
               currentIterator = inputIterators.get(counter++);
    +          // Initialize the new iterator
    +          currentIterator.initialize();
               hasNext = internalHasNext();
             }
           }
           return hasNext;
         }
     
    -    @Override
    -    public CarbonRowBatch next() {
    -      // Create batch and fill it.
    -      CarbonRowBatch carbonRowBatch = new CarbonRowBatch();
    -      int count = 0;
    -      while (internalHasNext() && count < batchSize) {
    -        carbonRowBatch.addRow(new CarbonRow(rowParser.parseRow(currentIterator.next())));
    -        count++;
    +    @Override public CarbonRowBatch next() {
    +      CarbonRowBatch result = null;
    +      try {
    +        if (future == null) {
    +          future = getCarbonRowBatch();
    +        }
    +        result = future.get();
    +        nextBatch = false;
    +        if (hasNext()) {
    +          nextBatch = true;
    +          future = getCarbonRowBatch();
    +        } else {
    +          currentIterator.close();
    +        }
    +      } catch (Exception e) {
    --- End diff --
   
    cache InterruptedException, ExecutionException only


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user jackylk commented on the issue:

    https://github.com/apache/incubator-carbondata/pull/333
 
    LGTM



---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[qiuchenjian-2]

Github user asfgit closed the pull request at:

    https://github.com/apache/incubator-carbondata/pull/333


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---