carbondata-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jackylk <...@git.apache.org>
Subject [GitHub] carbondata pull request #1642: [CARBONDATA-1855][PARTITION] Added outputform...
Date Fri, 15 Dec 2017 07:43:56 GMT
Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/1642#discussion_r157138950
  
    --- Diff: processing/src/main/java/org/apache/carbondata/processing/loading/iterator/CarbonOutputIteratorWrapper.java
---
    @@ -0,0 +1,118 @@
    +/*
    + * 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.loading.iterator;
    +
    +import java.util.concurrent.ArrayBlockingQueue;
    +
    +import org.apache.carbondata.common.CarbonIterator;
    +
    +public class CarbonOutputIteratorWrapper extends CarbonIterator<String[]> {
    +
    +  private boolean close = false;
    +
    +  /**
    +   * Number of rows kept in memory at most will be batchSize * queue size
    +   */
    +  private int batchSize = 1000;
    +
    +  private RowBatch loadBatch = new RowBatch(batchSize);
    +
    +  private RowBatch readBatch;
    +
    +  private ArrayBlockingQueue<RowBatch> queue = new ArrayBlockingQueue<>(10);
    +
    +  public void write(String[] row) throws InterruptedException {
    +
    +    if (!loadBatch.addRow(row)) {
    +      loadBatch.readyRead();
    +      queue.put(loadBatch);
    +      loadBatch = new RowBatch(batchSize);
    +    }
    +  }
    +
    +  @Override public boolean hasNext() {
    +    return !queue.isEmpty() || !close || readBatch != null && readBatch.hasNext();
    +  }
    +
    +  @Override public String[] next() {
    +    if (readBatch == null || !readBatch.hasNext()) {
    +      try {
    +        readBatch = queue.take();
    +      } catch (InterruptedException e) {
    +        throw new RuntimeException(e);
    +      }
    +    }
    +    return readBatch.next();
    +  }
    +
    +  @Override public void close() {
    +    if (loadBatch.isLoading()) {
    +      try {
    +        loadBatch.readyRead();
    +        queue.put(loadBatch);
    +      } catch (InterruptedException e) {
    +        throw new RuntimeException(e);
    +      }
    +    }
    +    close = true;
    +  }
    +
    +  private static class RowBatch extends CarbonIterator<String[]> {
    +
    +    private int counter;
    +
    +    private String[][] batch;
    +
    +    private int size;
    +
    +    private boolean isLoading = true;
    +
    +    public RowBatch(int size) {
    +      batch = new String[size][];
    +      this.size = size;
    +    }
    +
    +    public boolean addRow(String[] row) {
    --- End diff --
    
    Can you add comment, what does return value mean


---

Mime
View raw message