carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jack...@apache.org
Subject [45/50] [abbrv] incubator-carbondata git commit: no sort datawriter
Date Tue, 02 May 2017 14:13:09 GMT
no sort datawriter


Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/49cabeca
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/49cabeca
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/49cabeca

Branch: refs/heads/12-dev
Commit: 49cabeca67622486f0331757f3746728133d4f3d
Parents: 7dafae7
Author: QiangCai <qiangcai@qq.com>
Authored: Thu Apr 6 00:09:06 2017 +0530
Committer: jackylk <jacky.likun@huawei.com>
Committed: Tue May 2 09:42:20 2017 +0800

----------------------------------------------------------------------
 .../processing/merger/CarbonDataMergerUtil.java |   2 +-
 .../processing/model/CarbonLoadModel.java       |   6 +-
 .../newflow/DataLoadProcessBuilder.java         |  38 ++-
 .../CarbonRowDataWriterProcessorStepImpl.java   | 321 +++++++++++++++++++
 .../newflow/steps/NoSortProcessorStepImpl.java  | 153 ---------
 .../store/CarbonDataFileAttributes.java         |  10 +-
 .../store/CarbonFactDataHandlerModel.java       |   2 +-
 .../store/writer/AbstractFactDataWriter.java    |   4 +-
 8 files changed, 361 insertions(+), 175 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/49cabeca/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtil.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtil.java
b/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtil.java
index 65ec4fc..b9ea447 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtil.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtil.java
@@ -170,7 +170,7 @@ public final class CarbonDataMergerUtil {
     boolean updateLockStatus = false;
     boolean tableLockStatus = false;
 
-    String timestamp = carbonLoadModel.getFactTimeStamp();
+    String timestamp = "" + carbonLoadModel.getFactTimeStamp();
 
     List<String> updatedDeltaFilesList =
         new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/49cabeca/processing/src/main/java/org/apache/carbondata/processing/model/CarbonLoadModel.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/model/CarbonLoadModel.java
b/processing/src/main/java/org/apache/carbondata/processing/model/CarbonLoadModel.java
index d8f84bf..f63cb03 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/model/CarbonLoadModel.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/model/CarbonLoadModel.java
@@ -82,7 +82,7 @@ public class CarbonLoadModel implements Serializable {
   /**
    * new load start time
    */
-  private String factTimeStamp;
+  private long factTimeStamp;
   /**
    * load Id
    */
@@ -588,7 +588,7 @@ public class CarbonLoadModel implements Serializable {
   /**
    * @return
    */
-  public String getFactTimeStamp() {
+  public long getFactTimeStamp() {
     return factTimeStamp;
   }
 
@@ -596,7 +596,7 @@ public class CarbonLoadModel implements Serializable {
    * @param factTimeStamp
    */
   public void setFactTimeStamp(long factTimeStamp) {
-    this.factTimeStamp = factTimeStamp + "";
+    this.factTimeStamp = factTimeStamp;
   }
 
   public String[] getDelimiters() {

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/49cabeca/processing/src/main/java/org/apache/carbondata/processing/newflow/DataLoadProcessBuilder.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/newflow/DataLoadProcessBuilder.java
b/processing/src/main/java/org/apache/carbondata/processing/newflow/DataLoadProcessBuilder.java
index 11e5db1..de469f9 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/newflow/DataLoadProcessBuilder.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/newflow/DataLoadProcessBuilder.java
@@ -35,12 +35,12 @@ import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure;
 import org.apache.carbondata.core.util.CarbonProperties;
 import org.apache.carbondata.processing.model.CarbonLoadModel;
 import org.apache.carbondata.processing.newflow.constants.DataLoadProcessorConstants;
+import org.apache.carbondata.processing.newflow.steps.CarbonRowDataWriterProcessorStepImpl;
 import org.apache.carbondata.processing.newflow.steps.DataConverterProcessorStepImpl;
 import org.apache.carbondata.processing.newflow.steps.DataConverterProcessorWithBucketingStepImpl;
 import org.apache.carbondata.processing.newflow.steps.DataWriterBatchProcessorStepImpl;
 import org.apache.carbondata.processing.newflow.steps.DataWriterProcessorStepImpl;
 import org.apache.carbondata.processing.newflow.steps.InputProcessorStepImpl;
-import org.apache.carbondata.processing.newflow.steps.NoSortProcessorStepImpl;
 import org.apache.carbondata.processing.newflow.steps.SortProcessorStepImpl;
 import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
 
@@ -59,7 +59,9 @@ public final class DataLoadProcessBuilder {
             CarbonCommonConstants.LOAD_USE_BATCH_SORT_DEFAULT));
     CarbonDataLoadConfiguration configuration =
         createConfiguration(loadModel, storeLocation);
-    if (configuration.getBucketingInfo() != null) {
+    if (!configuration.isSortTable()) {
+      return buildInternalForNoSort(inputIterators, configuration);
+    } else if (configuration.getBucketingInfo() != null) {
       return buildInternalForBucketing(inputIterators, configuration);
     } else if (batchSort) {
       return buildInternalForBatchSort(inputIterators, configuration);
@@ -77,16 +79,30 @@ public final class DataLoadProcessBuilder {
     // data types and configurations.
     AbstractDataLoadProcessorStep converterProcessorStep =
         new DataConverterProcessorStepImpl(configuration, inputProcessorStep);
-    // 3. Sorts the data by SortColumn or not
-    AbstractDataLoadProcessorStep sortProcessorStep = configuration.isSortTable() ?
-        new SortProcessorStepImpl(configuration, converterProcessorStep) :
-        new NoSortProcessorStepImpl(configuration, converterProcessorStep);
+    // 3. Sorts the data by SortColumn
+    AbstractDataLoadProcessorStep sortProcessorStep =
+        new SortProcessorStepImpl(configuration, converterProcessorStep);
     // 4. Writes the sorted data in carbondata format.
     AbstractDataLoadProcessorStep writerProcessorStep =
         new DataWriterProcessorStepImpl(configuration, sortProcessorStep);
     return writerProcessorStep;
   }
 
+  private AbstractDataLoadProcessorStep buildInternalForNoSort(CarbonIterator[] inputIterators,
+      CarbonDataLoadConfiguration configuration) {
+    // 1. Reads the data input iterators and parses the data.
+    AbstractDataLoadProcessorStep inputProcessorStep =
+        new InputProcessorStepImpl(configuration, inputIterators);
+    // 2. Converts the data like dictionary or non dictionary or complex objects depends
on
+    // data types and configurations.
+    AbstractDataLoadProcessorStep converterProcessorStep =
+        new DataConverterProcessorStepImpl(configuration, inputProcessorStep);
+    // 3. Writes the sorted data in carbondata format.
+    AbstractDataLoadProcessorStep writerProcessorStep =
+        new CarbonRowDataWriterProcessorStepImpl(configuration, converterProcessorStep);
+    return writerProcessorStep;
+  }
+
   private AbstractDataLoadProcessorStep buildInternalForBatchSort(CarbonIterator[] inputIterators,
       CarbonDataLoadConfiguration configuration) {
     // 1. Reads the data input iterators and parses the data.
@@ -97,9 +113,8 @@ public final class DataLoadProcessBuilder {
     AbstractDataLoadProcessorStep converterProcessorStep =
         new DataConverterProcessorStepImpl(configuration, inputProcessorStep);
     // 3. Sorts the data by SortColumn or not
-    AbstractDataLoadProcessorStep sortProcessorStep = configuration.isSortTable() ?
-        new SortProcessorStepImpl(configuration, converterProcessorStep) :
-        new NoSortProcessorStepImpl(configuration, converterProcessorStep);
+    AbstractDataLoadProcessorStep sortProcessorStep =
+        new SortProcessorStepImpl(configuration, converterProcessorStep);
     // 4. Writes the sorted data in carbondata format.
     AbstractDataLoadProcessorStep writerProcessorStep =
         new DataWriterBatchProcessorStepImpl(configuration, sortProcessorStep);
@@ -116,9 +131,8 @@ public final class DataLoadProcessBuilder {
     AbstractDataLoadProcessorStep converterProcessorStep =
         new DataConverterProcessorWithBucketingStepImpl(configuration, inputProcessorStep);
     // 3. Sorts the data by SortColumn or not
-    AbstractDataLoadProcessorStep sortProcessorStep = configuration.isSortTable() ?
-        new SortProcessorStepImpl(configuration, converterProcessorStep) :
-        new NoSortProcessorStepImpl(configuration, converterProcessorStep);
+    AbstractDataLoadProcessorStep sortProcessorStep =
+        new SortProcessorStepImpl(configuration, converterProcessorStep);
     // 4. Writes the sorted data in carbondata format.
     AbstractDataLoadProcessorStep writerProcessorStep =
         new DataWriterProcessorStepImpl(configuration, sortProcessorStep);

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/49cabeca/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/CarbonRowDataWriterProcessorStepImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/CarbonRowDataWriterProcessorStepImpl.java
b/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/CarbonRowDataWriterProcessorStepImpl.java
new file mode 100644
index 0000000..929a09e
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/CarbonRowDataWriterProcessorStepImpl.java
@@ -0,0 +1,321 @@
+/*
+ * 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.steps;
+
+import java.io.File;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.util.Iterator;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datastore.block.SegmentProperties;
+import org.apache.carbondata.core.keygenerator.KeyGenException;
+import org.apache.carbondata.core.keygenerator.KeyGenerator;
+import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
+import org.apache.carbondata.core.util.CarbonTimeStatisticsFactory;
+import org.apache.carbondata.core.util.DataTypeUtil;
+import org.apache.carbondata.processing.newflow.AbstractDataLoadProcessorStep;
+import org.apache.carbondata.processing.newflow.CarbonDataLoadConfiguration;
+import org.apache.carbondata.processing.newflow.DataField;
+import org.apache.carbondata.processing.newflow.exception.CarbonDataLoadingException;
+import org.apache.carbondata.processing.newflow.row.CarbonRow;
+import org.apache.carbondata.processing.newflow.row.CarbonRowBatch;
+import org.apache.carbondata.processing.store.CarbonFactDataHandlerModel;
+import org.apache.carbondata.processing.store.CarbonFactHandler;
+import org.apache.carbondata.processing.store.CarbonFactHandlerFactory;
+import org.apache.carbondata.processing.store.writer.exception.CarbonDataWriterException;
+import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
+
+/**
+ * It reads data from sorted files which are generated in previous sort step.
+ * And it writes data to carbondata file. It also generates mdk key while writing to carbondata
file
+ */
+public class CarbonRowDataWriterProcessorStepImpl extends AbstractDataLoadProcessorStep {
+
+  private static final LogService LOGGER =
+      LogServiceFactory.getLogService(CarbonRowDataWriterProcessorStepImpl.class.getName());
+
+  private SegmentProperties segmentProperties;
+
+  private KeyGenerator keyGenerator;
+
+  private int dimensionWithComplexCount;
+
+  private int noDictWithComplextCount;
+
+  private boolean[] isNoDictionaryDimensionColumn;
+
+  private char[] aggType;
+
+  private int dimensionCount;
+
+  private int measureCount;
+
+  private long[] readCounter;
+
+  private long[] writeCounter;
+
+  private int outputLength;
+
+  private CarbonTableIdentifier tableIdentifier;
+
+  private String tableName;
+
+  public CarbonRowDataWriterProcessorStepImpl(CarbonDataLoadConfiguration configuration,
+      AbstractDataLoadProcessorStep child) {
+    super(configuration, child);
+  }
+
+  @Override public DataField[] getOutput() {
+    return child.getOutput();
+  }
+
+  @Override public void initialize() throws IOException {
+    child.initialize();
+  }
+
+  private String getStoreLocation(CarbonTableIdentifier tableIdentifier, String partitionId)
{
+    String storeLocation = CarbonDataProcessorUtil
+        .getLocalDataFolderLocation(tableIdentifier.getDatabaseName(),
+            tableIdentifier.getTableName(), String.valueOf(configuration.getTaskNo()), partitionId,
+            configuration.getSegmentId() + "", false);
+    new File(storeLocation).mkdirs();
+    return storeLocation;
+  }
+
+  @Override public Iterator<CarbonRowBatch>[] execute() throws CarbonDataLoadingException
{
+    final Iterator<CarbonRowBatch>[] iterators = child.execute();
+    tableIdentifier = configuration.getTableIdentifier().getCarbonTableIdentifier();
+    tableName = tableIdentifier.getTableName();
+    try {
+      readCounter = new long[iterators.length];
+      writeCounter = new long[iterators.length];
+      dimensionWithComplexCount = configuration.getDimensionCount();
+      noDictWithComplextCount =
+          configuration.getNoDictionaryCount() + configuration.getComplexDimensionCount();
+      dimensionCount = configuration.getDimensionCount() - noDictWithComplextCount;
+      isNoDictionaryDimensionColumn =
+          CarbonDataProcessorUtil.getNoDictionaryMapping(configuration.getDataFields());
+      aggType = CarbonDataProcessorUtil
+          .getAggType(configuration.getMeasureCount(), configuration.getMeasureFields());
+
+      CarbonFactDataHandlerModel dataHandlerModel = CarbonFactDataHandlerModel
+          .createCarbonFactDataHandlerModel(configuration,
+              getStoreLocation(tableIdentifier, String.valueOf(0)), 0, 0);
+      measureCount = dataHandlerModel.getMeasureCount();
+      outputLength = measureCount + (this.noDictWithComplextCount > 0 ? 1 : 0) + 1;
+      segmentProperties = dataHandlerModel.getSegmentProperties();
+      keyGenerator = segmentProperties.getDimensionKeyGenerator();
+
+      CarbonTimeStatisticsFactory.getLoadStatisticsInstance()
+          .recordDictionaryValue2MdkAdd2FileTime(configuration.getPartitionId(),
+              System.currentTimeMillis());
+
+      if (iterators.length == 1) {
+        doExecute(iterators[0], 0, 0);
+      } else {
+        ExecutorService executorService = Executors.newFixedThreadPool(iterators.length);
+        Future[] futures = new Future[iterators.length];
+        for (int i = 0; i < iterators.length; i++) {
+          futures[i] = executorService.submit(new DataWriterRunnable(iterators[i], i));
+        }
+        for (Future future : futures) {
+          future.get();
+        }
+      }
+    } catch (CarbonDataWriterException e) {
+      LOGGER.error(e, "Failed for table: " + tableName + " in DataWriterProcessorStepImpl");
+      throw new CarbonDataLoadingException(
+          "Error while initializing data handler : " + e.getMessage());
+    } catch (Exception e) {
+      LOGGER.error(e, "Failed for table: " + tableName + " in DataWriterProcessorStepImpl");
+      throw new CarbonDataLoadingException("There is an unexpected error: " + e.getMessage(),
e);
+    }
+    return null;
+  }
+
+  private void doExecute(Iterator<CarbonRowBatch> iterator, int partitionId, int iteratorIndex)
{
+    String storeLocation = getStoreLocation(tableIdentifier, String.valueOf(partitionId));
+    CarbonFactDataHandlerModel model = CarbonFactDataHandlerModel
+        .createCarbonFactDataHandlerModel(configuration, storeLocation, partitionId, 0);
+    model.getCarbonDataFileAttributes()
+        .setFactTimeStamp(model.getCarbonDataFileAttributes().getFactTimeStamp() + iteratorIndex);
+    CarbonFactHandler dataHandler = null;
+    boolean rowsNotExist = true;
+    while (iterator.hasNext()) {
+      if (rowsNotExist) {
+        rowsNotExist = false;
+        dataHandler = CarbonFactHandlerFactory
+            .createCarbonFactHandler(model, CarbonFactHandlerFactory.FactHandlerType.COLUMNAR);
+        dataHandler.initialise();
+      }
+      processBatch(iterator.next(), dataHandler, iteratorIndex);
+    }
+    if (!rowsNotExist) {
+      finish(dataHandler, iteratorIndex);
+    }
+  }
+
+  @Override protected String getStepName() {
+    return "Data Writer";
+  }
+
+  private void finish(CarbonFactHandler dataHandler, int iteratorIndex) {
+    try {
+      dataHandler.finish();
+    } catch (Exception e) {
+      LOGGER.error(e, "Failed for table: " + tableName + " in  finishing data handler");
+    }
+    LOGGER.info("Record Processed For table: " + tableName);
+    String logMessage =
+        "Finished Carbon DataWriterProcessorStepImpl: Read: " + readCounter[iteratorIndex]
+            + ": Write: " + readCounter[iteratorIndex];
+    LOGGER.info(logMessage);
+    CarbonTimeStatisticsFactory.getLoadStatisticsInstance().recordTotalRecords(rowCounter.get());
+    processingComplete(dataHandler);
+    CarbonTimeStatisticsFactory.getLoadStatisticsInstance()
+        .recordDictionaryValue2MdkAdd2FileTime(configuration.getPartitionId(),
+            System.currentTimeMillis());
+    CarbonTimeStatisticsFactory.getLoadStatisticsInstance()
+        .recordMdkGenerateTotalTime(configuration.getPartitionId(), System.currentTimeMillis());
+  }
+
+  private void processingComplete(CarbonFactHandler dataHandler) throws CarbonDataLoadingException
{
+    if (null != dataHandler) {
+      try {
+        dataHandler.closeHandler();
+      } catch (CarbonDataWriterException e) {
+        LOGGER.error(e, e.getMessage());
+        throw new CarbonDataLoadingException(e.getMessage());
+      } catch (Exception e) {
+        LOGGER.error(e, e.getMessage());
+        throw new CarbonDataLoadingException("There is an unexpected error: " + e.getMessage());
+      }
+    }
+  }
+
+  /**
+   * convert input CarbonRow to output CarbonRow
+   * e.g. There is a table as following,
+   * the number of dictionary dimensions is a,
+   * the number of no-dictionary dimensions is b,
+   * the number of complex dimensions is c,
+   * the number of measures is d.
+   * input CarbonRow format:  the length of Object[] data is a+b+c+d, the number of all columns.
+   * ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+   * | Part                     | Object item                    | describe             
   |
+   * ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+   * | Object[0 ~ a+b-1]        | Integer, byte[], Integer, ...  | dict + no dict dimensions|
+   * ----------------------------------------------------------------------------------------
+   * | Object[a+b ~ a+b+c-1]    | byte[], byte[], ...            | complex dimensions   
   |
+   * ----------------------------------------------------------------------------------------
+   * | Object[a+b+c ~ a+b+c+d-1]| int, byte[], ...               | measures             
   |
+   * ----------------------------------------------------------------------------------------
+   * output CarbonRow format: the length of object[] data is d + (b+c>0?1:0) + 1.
+   * ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+   * | Part                     | Object item                    | describe             
   |
+   * ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+   * | Object[0 ~ d-1]          | int, byte[], ...               | measures             
   |
+   * ----------------------------------------------------------------------------------------
+   * | Object[d]                | byte[b+c][]                    | no dict + complex dim
   |
+   * ----------------------------------------------------------------------------------------
+   * | Object[d+1]              | byte[]                         | mdkey                
   |
+   * ----------------------------------------------------------------------------------------
+   *
+   * @param row
+   * @return
+   */
+  private Object[] convertRow(CarbonRow row) throws KeyGenException {
+    int dictIndex = 0;
+    int nonDicIndex = 0;
+    int[] dim = new int[this.dimensionCount];
+    byte[][] nonDicArray = new byte[this.noDictWithComplextCount][];
+    // read dimension values
+    int dimCount = 0;
+    for (; dimCount < isNoDictionaryDimensionColumn.length; dimCount++) {
+      if (isNoDictionaryDimensionColumn[dimCount]) {
+        nonDicArray[nonDicIndex++] = (byte[]) row.getObject(dimCount);
+      } else {
+        dim[dictIndex++] = (int) row.getObject(dimCount);
+      }
+    }
+
+    for (; dimCount < this.dimensionWithComplexCount; dimCount++) {
+      nonDicArray[nonDicIndex++] = (byte[]) row.getObject(dimCount);
+    }
+
+    int l = 0;
+    Object[] outputRow = new Object[outputLength];
+    for (; l < this.measureCount; l++) {
+      Object value = row.getObject(l + this.dimensionWithComplexCount);
+      if (null != value) {
+        if (aggType[l] == CarbonCommonConstants.BIG_DECIMAL_MEASURE) {
+          BigDecimal val = (BigDecimal) value;
+          outputRow[l] = DataTypeUtil.bigDecimalToByte(val);
+        } else {
+          outputRow[l] = value;
+        }
+      } else {
+        outputRow[l] = null;
+      }
+    }
+
+    if (this.noDictWithComplextCount > 0) {
+      outputRow[l++] = nonDicArray;
+    }
+    outputRow[l] = keyGenerator.generateKey(dim);
+    return outputRow;
+  }
+
+  private void processBatch(CarbonRowBatch batch, CarbonFactHandler dataHandler, int iteratorIndex)
+      throws CarbonDataLoadingException {
+    try {
+      while (batch.hasNext()) {
+        dataHandler.addDataToStore(convertRow(batch.next()));
+        readCounter[iteratorIndex]++;
+      }
+      writeCounter[iteratorIndex] += batch.getSize();
+    } catch (Exception e) {
+      throw new CarbonDataLoadingException("unable to generate the mdkey", e);
+    }
+    rowCounter.getAndAdd(batch.getSize());
+  }
+
+  @Override protected CarbonRow processRow(CarbonRow row) {
+    return null;
+  }
+
+  class DataWriterRunnable implements Runnable {
+
+    private Iterator<CarbonRowBatch> iterator;
+    private int iteratorIndex = 0;
+
+    DataWriterRunnable(Iterator<CarbonRowBatch> iterator, int iteratorIndex) {
+      this.iterator = iterator;
+      this.iteratorIndex = iteratorIndex;
+    }
+
+    @Override public void run() {
+      doExecute(this.iterator, 0, iteratorIndex);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/49cabeca/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/NoSortProcessorStepImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/NoSortProcessorStepImpl.java
b/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/NoSortProcessorStepImpl.java
deleted file mode 100644
index bde89ed..0000000
--- a/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/NoSortProcessorStepImpl.java
+++ /dev/null
@@ -1,153 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.carbondata.processing.newflow.steps;
-
-import java.io.IOException;
-import java.math.BigDecimal;
-
-import org.apache.carbondata.core.constants.CarbonCommonConstants;
-import org.apache.carbondata.core.util.DataTypeUtil;
-import org.apache.carbondata.processing.newflow.AbstractDataLoadProcessorStep;
-import org.apache.carbondata.processing.newflow.CarbonDataLoadConfiguration;
-import org.apache.carbondata.processing.newflow.DataField;
-import org.apache.carbondata.processing.newflow.row.CarbonRow;
-import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
-import org.apache.carbondata.processing.util.NonDictionaryUtil;
-
-/**
- * if the table doesn't have sort_columns, just convert row format.
- */
-public class NoSortProcessorStepImpl extends AbstractDataLoadProcessorStep {
-
-  private int dimensionCount;
-
-  private int dimensionWithComplexCount;
-
-  private int noDictCount;
-
-  private int measureCount;
-
-  private boolean[] isNoDictionaryDimensionColumn;
-
-  private char[] aggType;
-
-  public NoSortProcessorStepImpl(CarbonDataLoadConfiguration configuration,
-      AbstractDataLoadProcessorStep child) {
-    super(configuration, child);
-    this.dimensionWithComplexCount = configuration.getDimensionCount();
-    this.noDictCount =
-        configuration.getNoDictionaryCount() + configuration.getComplexDimensionCount();
-    this.dimensionCount = configuration.getDimensionCount() - this.noDictCount;
-    this.measureCount = configuration.getMeasureCount();
-    this.isNoDictionaryDimensionColumn =
-        CarbonDataProcessorUtil.getNoDictionaryMapping(configuration.getDataFields());
-    this.aggType = CarbonDataProcessorUtil
-        .getAggType(configuration.getMeasureCount(), configuration.getMeasureFields());
-  }
-
-  @Override public DataField[] getOutput() {
-    return child.getOutput();
-  }
-
-  @Override public void initialize() throws IOException {
-    child.initialize();
-  }
-
-  /**
-   * convert input CarbonRow to output CarbonRow
-   * e.g. There is a table as following,
-   * the number of dictionary dimensions is a,
-   * the number of no-dictionary dimensions is b,
-   * the number of complex dimensions is c,
-   * the number of measures is d.
-   * input CarbonRow format:  the length of Object[] data is a+b+c+d, the number of all columns.
-   * ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
-   * | Part                     | Object item                    | describe             
   |
-   * ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
-   * | Object[0 ~ a+b-1]        | Integer, byte[], Integer, ...  | dict + no dict dimensions|
-   * ----------------------------------------------------------------------------------------
-   * | Object[a+b ~ a+b+c-1]    | byte[], byte[], ...            | complex dimensions   
   |
-   * ----------------------------------------------------------------------------------------
-   * | Object[a+b+c ~ a+b+c+d-1]| int, byte[], ...               | measures             
   |
-   * ----------------------------------------------------------------------------------------
-   * output CarbonRow format: the length of object[] data is 3.
-   * ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
-   * | Part                     | Object item                    | describe             
   |
-   * ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
-   * | Object[0]                | int[a]                         | dict dimension array 
   |
-   * ----------------------------------------------------------------------------------------
-   * | Object[1]                | byte[b+c][]                    | no dict + complex dim
   |
-   * ----------------------------------------------------------------------------------------
-   * | Object[2]                | Object[d]                      | measures             
   |
-   * ----------------------------------------------------------------------------------------
-   * @param row
-   * @return
-   */
-  @Override protected CarbonRow processRow(CarbonRow row) {
-    int dictIndex = 0;
-    int nonDicIndex = 0;
-    int[] dim = new int[this.dimensionCount];
-    byte[][] nonDicArray = new byte[this.noDictCount][];
-    Object[] measures = new Object[this.measureCount];
-    // read dimension values
-    int dimCount = 0;
-    for (; dimCount < isNoDictionaryDimensionColumn.length; dimCount++) {
-      if (isNoDictionaryDimensionColumn[dimCount]) {
-        nonDicArray[nonDicIndex++] = (byte[]) row.getObject(dimCount);
-      } else {
-        dim[dictIndex++] = (int) row.getObject(dimCount);
-      }
-    }
-
-    for (; dimCount < this.dimensionWithComplexCount; dimCount++) {
-      nonDicArray[nonDicIndex++] = (byte[]) row.getObject(dimCount);
-    }
-
-    // measure values
-    for (int mesCount = 0; mesCount < this.measureCount; mesCount++) {
-      Object value = row.getObject(mesCount + this.dimensionWithComplexCount);
-      if (null != value) {
-        if (aggType[mesCount] == CarbonCommonConstants.DOUBLE_MEASURE) {
-          measures[mesCount] = value;
-        } else if (aggType[mesCount] == CarbonCommonConstants.BIG_INT_MEASURE) {
-          measures[mesCount] = value;
-        } else if (aggType[mesCount] == CarbonCommonConstants.BIG_DECIMAL_MEASURE) {
-          BigDecimal val = (BigDecimal) value;
-          measures[mesCount] = DataTypeUtil.bigDecimalToByte(val);
-        }
-      } else {
-        measures[mesCount] = null;
-      }
-    }
-    // create new row of size 3 (1 for dims , 1 for high card , 1 for measures)
-    Object[] holder = new Object[3];
-    NonDictionaryUtil.prepareOutObj(holder, dim, nonDicArray, measures);
-    //return out row
-    return new CarbonRow(holder);
-  }
-
-  @Override
-  public void close() {
-    if (!closed) {
-      super.close();
-    }
-  }
-
-  @Override protected String getStepName() {
-    return "No Sort Processor";
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/49cabeca/processing/src/main/java/org/apache/carbondata/processing/store/CarbonDataFileAttributes.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonDataFileAttributes.java
b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonDataFileAttributes.java
index 188c468..0b606b0 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonDataFileAttributes.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonDataFileAttributes.java
@@ -39,13 +39,13 @@ public class CarbonDataFileAttributes {
   /**
    * load start time
    */
-  private String factTimeStamp;
+  private long factTimeStamp;
 
   /**
    * @param taskId
    * @param factTimeStamp
    */
-  public CarbonDataFileAttributes(int taskId, String factTimeStamp) {
+  public CarbonDataFileAttributes(int taskId, long factTimeStamp) {
     this.taskId = taskId;
     this.factTimeStamp = factTimeStamp;
   }
@@ -57,10 +57,14 @@ public class CarbonDataFileAttributes {
     return taskId;
   }
 
+  public void setFactTimeStamp(long factTimeStamp) {
+    this.factTimeStamp = factTimeStamp;
+  }
+
   /**
    * @return fact time stamp which is load start time
    */
-  public String getFactTimeStamp() {
+  public long getFactTimeStamp() {
     return factTimeStamp;
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/49cabeca/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
index d3b5605..61322bc 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
@@ -265,7 +265,7 @@ public class CarbonFactDataHandlerModel {
 
     CarbonDataFileAttributes carbonDataFileAttributes =
         new CarbonDataFileAttributes(Integer.parseInt(configuration.getTaskNo()),
-            (String) configuration.getDataLoadProperty(DataLoadProcessorConstants.FACT_TIME_STAMP));
+            (Long) configuration.getDataLoadProperty(DataLoadProcessorConstants.FACT_TIME_STAMP));
     String carbonDataDirectoryPath = getCarbonDataFolderLocation(configuration);
 
     CarbonFactDataHandlerModel carbonFactDataHandlerModel = new CarbonFactDataHandlerModel();

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/49cabeca/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
b/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
index cda907c..770d24c 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
@@ -296,7 +296,7 @@ public abstract class AbstractFactDataWriter<T> implements CarbonFactDataWriter<
     this.carbonDataFileName = carbonTablePath
         .getCarbonDataFileName(fileCount, dataWriterVo.getCarbonDataFileAttributes().getTaskId(),
             dataWriterVo.getBucketNumber(), dataWriterVo.getTaskExtension(),
-            dataWriterVo.getCarbonDataFileAttributes().getFactTimeStamp());
+            "" + dataWriterVo.getCarbonDataFileAttributes().getFactTimeStamp());
     String actualFileNameVal = carbonDataFileName + CarbonCommonConstants.FILE_INPROGRESS_STATUS;
     FileData fileData = new FileData(actualFileNameVal, dataWriterVo.getStoreLocation());
     dataWriterVo.getFileManager().add(fileData);
@@ -447,7 +447,7 @@ public abstract class AbstractFactDataWriter<T> implements CarbonFactDataWriter<
     String fileName = dataWriterVo.getStoreLocation() + File.separator + carbonTablePath
         .getCarbonIndexFileName(dataWriterVo.getCarbonDataFileAttributes().getTaskId(),
             dataWriterVo.getBucketNumber(), dataWriterVo.getTaskExtension(),
-            dataWriterVo.getCarbonDataFileAttributes().getFactTimeStamp());
+            "" + dataWriterVo.getCarbonDataFileAttributes().getFactTimeStamp());
     CarbonIndexFileWriter writer = new CarbonIndexFileWriter();
     // open file
     writer.openThriftWriter(fileName);


Mime
View raw message