carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ravipes...@apache.org
Subject [1/2] carbondata git commit: [CARBONDATA-2443][SDK]Multi level complex type support for AVRO based SDK
Date Tue, 08 May 2018 11:20:57 GMT
Repository: carbondata
Updated Branches:
  refs/heads/master 51db049c4 -> 74ea24d14


[CARBONDATA-2443][SDK]Multi level complex type support for AVRO based SDK

Multi level complex type support for AVRO based SDK

This closes #2276


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

Branch: refs/heads/master
Commit: ec33c11286ebe8009ac07698bf23ffb3bd3e7711
Parents: 51db049
Author: sounakr <sounakr@gmail.com>
Authored: Mon May 7 06:51:54 2018 +0530
Committer: ravipesala <ravi.pesala@gmail.com>
Committed: Tue May 8 16:49:08 2018 +0530

----------------------------------------------------------------------
 .../schema/table/TableSchemaBuilder.java        |  17 +-
 .../TestNonTransactionalCarbonTable.scala       |  10 +-
 .../datasources/SparkCarbonTableFormat.scala    |   2 +-
 .../loading/DataLoadProcessBuilder.java         |  12 +-
 .../converter/impl/FieldEncoderFactory.java     |   2 +-
 .../loading/model/CarbonLoadModel.java          |  14 +-
 .../InputProcessorStepForPartitionImpl.java     | 251 ---------------
 .../InputProcessorStepWithNoConverterImpl.java  | 306 +++++++++++++++++++
 .../util/CarbonDataProcessorUtil.java           |  29 ++
 .../carbondata/sdk/file/AvroCarbonWriter.java   |  60 ++--
 .../sdk/file/CarbonWriterBuilder.java           |   6 +
 11 files changed, 401 insertions(+), 308 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/carbondata/blob/ec33c112/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchemaBuilder.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchemaBuilder.java
b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchemaBuilder.java
index 42bb958..e3c07fa 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchemaBuilder.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchemaBuilder.java
@@ -184,7 +184,8 @@ public class TableSchemaBuilder {
     if (field.getDataType().isComplexType()) {
       String parentFieldName = newColumn.getColumnName();
       if (field.getDataType().getName().equalsIgnoreCase("ARRAY")) {
-        addColumn(new StructField("val",
+        String colName = getColNameForArray(parentFieldName);
+        addColumn(new StructField(colName,
             ((ArrayType) field.getDataType()).getElementType()), field.getFieldName(), false,
true);
       } else if (field.getDataType().getName().equalsIgnoreCase("STRUCT")
           && ((StructType) field.getDataType()).getFields().size() > 0) {
@@ -198,6 +199,20 @@ public class TableSchemaBuilder {
     return newColumn;
   }
 
+  private String getColNameForArray(String parentFieldName) {
+    if (!parentFieldName.contains(".val")) {
+      return "val";
+    } else {
+      String[] splits = parentFieldName.split("val");
+      if (splits.length == 1) {
+        return "val" + 1;
+      } else {
+        return "val" + (Integer.parseInt(parentFieldName
+            .substring(parentFieldName.lastIndexOf("val") + 3, parentFieldName.length()))
+ 1);
+      }
+    }
+  }
+
   /**
    * Throw exception if {@param field} name is repeated
    */

http://git-wip-us.apache.org/repos/asf/carbondata/blob/ec33c112/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestNonTransactionalCarbonTable.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestNonTransactionalCarbonTable.scala
b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestNonTransactionalCarbonTable.scala
index fabcd02..6b02d5a 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestNonTransactionalCarbonTable.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestNonTransactionalCarbonTable.scala
@@ -678,8 +678,8 @@ class TestNonTransactionalCarbonTable extends QueryTest with BeforeAndAfterAll
{
 
     val exception =
       intercept[RuntimeException] {
-      buildTestDataWithBadRecordFail()
-    }
+        buildTestDataWithBadRecordFail()
+      }
     assert(exception.getMessage()
       .contains("Data load failed due to bad record"))
 
@@ -780,7 +780,7 @@ class TestNonTransactionalCarbonTable extends QueryTest with BeforeAndAfterAll
{
                      |      }
                      |      ]
                      |  }
-                     """.stripMargin
+                   """.stripMargin
 
     val json: String = """ {"name": "bob","age": 10,"address": ["abc", "defg"]} """
 
@@ -835,8 +835,8 @@ class TestNonTransactionalCarbonTable extends QueryTest with BeforeAndAfterAll
{
 
     val json =
       """ {"name":"bob", "age":10,
-          |"address" : {"street":"abc", "city":"bang"},
-          |"doorNum" : [1,2,3,4]}""".stripMargin
+        |"address" : {"street":"abc", "city":"bang"},
+        |"doorNum" : [1,2,3,4]}""".stripMargin
 
     val fields = new Array[Field](4)
     fields(0) = new Field("name", DataTypes.STRING)

http://git-wip-us.apache.org/repos/asf/carbondata/blob/ec33c112/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala
index 1928b38..d6eab1d 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala
@@ -111,7 +111,7 @@ with Serializable {
     model.setDictionaryServerHost(options.getOrElse("dicthost", null))
     model.setDictionaryServerPort(options.getOrElse("dictport", "-1").toInt)
     CarbonTableOutputFormat.setOverwrite(conf, options("overwrite").toBoolean)
-    model.setPartitionLoad(true)
+    model.setLoadWithoutCoverterStep(true)
 
     val staticPartition = options.getOrElse("staticpartition", null)
     if (staticPartition != null) {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/ec33c112/processing/src/main/java/org/apache/carbondata/processing/loading/DataLoadProcessBuilder.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/DataLoadProcessBuilder.java
b/processing/src/main/java/org/apache/carbondata/processing/loading/DataLoadProcessBuilder.java
index 028c404..2f904ed 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/loading/DataLoadProcessBuilder.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/loading/DataLoadProcessBuilder.java
@@ -44,8 +44,8 @@ import org.apache.carbondata.processing.loading.steps.CarbonRowDataWriterProcess
 import org.apache.carbondata.processing.loading.steps.DataConverterProcessorStepImpl;
 import org.apache.carbondata.processing.loading.steps.DataWriterBatchProcessorStepImpl;
 import org.apache.carbondata.processing.loading.steps.DataWriterProcessorStepImpl;
-import org.apache.carbondata.processing.loading.steps.InputProcessorStepForPartitionImpl;
 import org.apache.carbondata.processing.loading.steps.InputProcessorStepImpl;
+import org.apache.carbondata.processing.loading.steps.InputProcessorStepWithNoConverterImpl;
 import org.apache.carbondata.processing.loading.steps.SortProcessorStepImpl;
 import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
 
@@ -62,8 +62,8 @@ public final class DataLoadProcessBuilder {
       CarbonIterator[] inputIterators) throws Exception {
     CarbonDataLoadConfiguration configuration = createConfiguration(loadModel, storeLocation);
     SortScopeOptions.SortScope sortScope = CarbonDataProcessorUtil.getSortScope(configuration);
-    if (loadModel.isPartitionLoad()) {
-      return buildInternalForPartitionLoad(inputIterators, configuration, sortScope);
+    if (loadModel.isLoadWithoutCoverterStep()) {
+      return buildInternalWithNoConverter(inputIterators, configuration, sortScope);
     } else if (!configuration.isSortTable() ||
         sortScope.equals(SortScopeOptions.SortScope.NO_SORT)) {
       return buildInternalForNoSort(inputIterators, configuration);
@@ -106,14 +106,14 @@ public final class DataLoadProcessBuilder {
   }
 
   /**
-   * Build pipe line for partition load
+   * Build pipe line for Load without Conversion Step.
    */
-  private AbstractDataLoadProcessorStep buildInternalForPartitionLoad(
+  private AbstractDataLoadProcessorStep buildInternalWithNoConverter(
       CarbonIterator[] inputIterators, CarbonDataLoadConfiguration configuration,
       SortScopeOptions.SortScope sortScope) {
     // Wraps with dummy processor.
     AbstractDataLoadProcessorStep inputProcessorStep =
-        new InputProcessorStepForPartitionImpl(configuration, inputIterators);
+        new InputProcessorStepWithNoConverterImpl(configuration, inputIterators);
     if (sortScope.equals(SortScopeOptions.SortScope.LOCAL_SORT)) {
       AbstractDataLoadProcessorStep sortProcessorStep =
           new SortProcessorStepImpl(configuration, inputProcessorStep);

http://git-wip-us.apache.org/repos/asf/carbondata/blob/ec33c112/processing/src/main/java/org/apache/carbondata/processing/loading/converter/impl/FieldEncoderFactory.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/converter/impl/FieldEncoderFactory.java
b/processing/src/main/java/org/apache/carbondata/processing/loading/converter/impl/FieldEncoderFactory.java
index 567a8b5..dd28dc6 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/loading/converter/impl/FieldEncoderFactory.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/loading/converter/impl/FieldEncoderFactory.java
@@ -128,7 +128,7 @@ public class FieldEncoderFactory {
   /**
    * Create parser for the carbon column.
    */
-  private static GenericDataType createComplexDataType(DataField dataField,
+  public static GenericDataType createComplexDataType(DataField dataField,
       Cache<DictionaryColumnUniqueIdentifier, Dictionary> cache,
       AbsoluteTableIdentifier absoluteTableIdentifier, DictionaryClient client, Boolean useOnePass,
       Map<Object, Integer> localCache, int index, String nullFormat, Boolean isEmptyBadRecords)
{

http://git-wip-us.apache.org/repos/asf/carbondata/blob/ec33c112/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModel.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModel.java
b/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModel.java
index 2a846e2..0cc0da3 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModel.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModel.java
@@ -201,8 +201,10 @@ public class CarbonLoadModel implements Serializable {
 
   /**
    * It directly writes data directly to nosort processor bypassing all other processors.
+   * For this method there will be no data conversion step. It writes data which is directly
+   * pushed into.
    */
-  private boolean isPartitionLoad;
+  private boolean isLoadWithoutCoverterStep;
 
   /**
    * Flder path to where data should be written for this load.
@@ -435,7 +437,7 @@ public class CarbonLoadModel implements Serializable {
     copy.batchSortSizeInMb = batchSortSizeInMb;
     copy.isAggLoadRequest = isAggLoadRequest;
     copy.badRecordsLocation = badRecordsLocation;
-    copy.isPartitionLoad = isPartitionLoad;
+    copy.isLoadWithoutCoverterStep = isLoadWithoutCoverterStep;
     copy.sortColumnsBoundsStr = sortColumnsBoundsStr;
     return copy;
   }
@@ -814,12 +816,12 @@ public class CarbonLoadModel implements Serializable {
   }
 
 
-  public boolean isPartitionLoad() {
-    return isPartitionLoad;
+  public boolean isLoadWithoutCoverterStep() {
+    return isLoadWithoutCoverterStep;
   }
 
-  public void setPartitionLoad(boolean partitionLoad) {
-    isPartitionLoad = partitionLoad;
+  public void setLoadWithoutCoverterStep(boolean loadWithoutCoverterStep) {
+    isLoadWithoutCoverterStep = loadWithoutCoverterStep;
   }
 
   public String getDataWritePath() {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/ec33c112/processing/src/main/java/org/apache/carbondata/processing/loading/steps/InputProcessorStepForPartitionImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/steps/InputProcessorStepForPartitionImpl.java
b/processing/src/main/java/org/apache/carbondata/processing/loading/steps/InputProcessorStepForPartitionImpl.java
deleted file mode 100644
index 1dc9b27..0000000
--- a/processing/src/main/java/org/apache/carbondata/processing/loading/steps/InputProcessorStepForPartitionImpl.java
+++ /dev/null
@@ -1,251 +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.loading.steps;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.carbondata.common.CarbonIterator;
-import org.apache.carbondata.core.datastore.row.CarbonRow;
-import org.apache.carbondata.core.metadata.datatype.DataType;
-import org.apache.carbondata.core.metadata.datatype.DataTypes;
-import org.apache.carbondata.core.metadata.encoder.Encoding;
-import org.apache.carbondata.core.util.CarbonProperties;
-import org.apache.carbondata.core.util.DataTypeUtil;
-import org.apache.carbondata.processing.loading.AbstractDataLoadProcessorStep;
-import org.apache.carbondata.processing.loading.CarbonDataLoadConfiguration;
-import org.apache.carbondata.processing.loading.DataField;
-import org.apache.carbondata.processing.loading.converter.impl.RowConverterImpl;
-import org.apache.carbondata.processing.loading.row.CarbonRowBatch;
-import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
-
-/**
- * It reads data from record reader and sends data to next step.
- */
-public class InputProcessorStepForPartitionImpl extends AbstractDataLoadProcessorStep {
-
-  private CarbonIterator<Object[]>[] inputIterators;
-
-  private boolean[] noDictionaryMapping;
-
-  private DataType[] dataTypes;
-
-  private int[] orderOfData;
-
-  public InputProcessorStepForPartitionImpl(CarbonDataLoadConfiguration configuration,
-      CarbonIterator<Object[]>[] inputIterators) {
-    super(configuration, null);
-    this.inputIterators = inputIterators;
-  }
-
-  @Override public DataField[] getOutput() {
-    return configuration.getDataFields();
-  }
-
-  @Override public void initialize() throws IOException {
-    super.initialize();
-    // if logger is enabled then raw data will be required.
-    RowConverterImpl rowConverter =
-        new RowConverterImpl(configuration.getDataFields(), configuration, null);
-    rowConverter.initialize();
-    configuration.setCardinalityFinder(rowConverter);
-    noDictionaryMapping =
-        CarbonDataProcessorUtil.getNoDictionaryMapping(configuration.getDataFields());
-    dataTypes = new DataType[configuration.getDataFields().length];
-    for (int i = 0; i < dataTypes.length; i++) {
-      if (configuration.getDataFields()[i].getColumn().hasEncoding(Encoding.DICTIONARY))
{
-        dataTypes[i] = DataTypes.INT;
-      } else {
-        dataTypes[i] = configuration.getDataFields()[i].getColumn().getDataType();
-      }
-    }
-    orderOfData = arrangeData(configuration.getDataFields(), configuration.getHeader());
-  }
-
-  private int[] arrangeData(DataField[] dataFields, String[] header) {
-    int[] data = new int[dataFields.length];
-    for (int i = 0; i < dataFields.length; i++) {
-      for (int j = 0; j < header.length; j++) {
-        if (dataFields[i].getColumn().getColName().equalsIgnoreCase(header[j])) {
-          data[i] = j;
-          break;
-        }
-      }
-    }
-    return data;
-  }
-
-  @Override public Iterator<CarbonRowBatch>[] execute() {
-    int batchSize = CarbonProperties.getInstance().getBatchSize();
-    List<CarbonIterator<Object[]>>[] readerIterators = partitionInputReaderIterators();
-    Iterator<CarbonRowBatch>[] outIterators = new Iterator[readerIterators.length];
-    for (int i = 0; i < outIterators.length; i++) {
-      outIterators[i] =
-          new InputProcessorIterator(readerIterators[i], batchSize, configuration.isPreFetch(),
-              rowCounter, orderOfData, noDictionaryMapping, dataTypes);
-    }
-    return outIterators;
-  }
-
-  /**
-   * Partition input iterators equally as per the number of threads.
-   *
-   * @return
-   */
-  private List<CarbonIterator<Object[]>>[] partitionInputReaderIterators() {
-    // Get the number of cores configured in property.
-    int numberOfCores = CarbonProperties.getInstance().getNumberOfCores();
-    // Get the minimum of number of cores and iterators size to get the number of parallel
threads
-    // to be launched.
-    int parallelThreadNumber = Math.min(inputIterators.length, numberOfCores);
-
-    List<CarbonIterator<Object[]>>[] iterators = new List[parallelThreadNumber];
-    for (int i = 0; i < parallelThreadNumber; i++) {
-      iterators[i] = new ArrayList<>();
-    }
-    // Equally partition the iterators as per number of threads
-    for (int i = 0; i < inputIterators.length; i++) {
-      iterators[i % parallelThreadNumber].add(inputIterators[i]);
-    }
-    return iterators;
-  }
-
-  @Override protected CarbonRow processRow(CarbonRow row) {
-    return null;
-  }
-
-  @Override public void close() {
-    if (!closed) {
-      super.close();
-      for (CarbonIterator inputIterator : inputIterators) {
-        inputIterator.close();
-      }
-    }
-  }
-
-  @Override protected String getStepName() {
-    return "Input Processor";
-  }
-
-  /**
-   * 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<CarbonIterator<Object[]>> inputIterators;
-
-    private CarbonIterator<Object[]> currentIterator;
-
-    private int counter;
-
-    private int batchSize;
-
-    private boolean nextBatch;
-
-    private boolean firstTime;
-
-    private AtomicLong rowCounter;
-
-    private boolean[] noDictionaryMapping;
-
-    private DataType[] dataTypes;
-
-    private int[] orderOfData;
-
-    public InputProcessorIterator(List<CarbonIterator<Object[]>> inputIterators,
int batchSize,
-        boolean preFetch, AtomicLong rowCounter, int[] orderOfData, boolean[] noDictionaryMapping,
-        DataType[] dataTypes) {
-      this.inputIterators = inputIterators;
-      this.batchSize = batchSize;
-      this.counter = 0;
-      // Get the first iterator from the list.
-      currentIterator = inputIterators.get(counter++);
-      this.rowCounter = rowCounter;
-      this.nextBatch = false;
-      this.firstTime = true;
-      this.noDictionaryMapping = noDictionaryMapping;
-      this.dataTypes = dataTypes;
-      this.orderOfData = orderOfData;
-    }
-
-    @Override public boolean hasNext() {
-      return nextBatch || internalHasNext();
-    }
-
-    private boolean internalHasNext() {
-      if (firstTime) {
-        firstTime = false;
-        currentIterator.initialize();
-      }
-      boolean hasNext = currentIterator.hasNext();
-      // If iterator is finished then check for next iterator.
-      if (!hasNext) {
-        currentIterator.close();
-        // Check next iterator is available in the list.
-        if (counter < inputIterators.size()) {
-          // 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() {
-      return getBatch();
-    }
-
-    private CarbonRowBatch getBatch() {
-      // Create batch and fill it.
-      CarbonRowBatch carbonRowBatch = new CarbonRowBatch(batchSize);
-      int count = 0;
-      while (internalHasNext() && count < batchSize) {
-        carbonRowBatch.addRow(new CarbonRow(convertToNoDictionaryToBytes(currentIterator.next())));
-        count++;
-      }
-      rowCounter.getAndAdd(carbonRowBatch.getSize());
-      return carbonRowBatch;
-    }
-
-    private Object[] convertToNoDictionaryToBytes(Object[] data) {
-      Object[] newData = new Object[data.length];
-      for (int i = 0; i < noDictionaryMapping.length; i++) {
-        if (noDictionaryMapping[i]) {
-          newData[i] = DataTypeUtil
-              .getBytesDataDataTypeForNoDictionaryColumn(data[orderOfData[i]], dataTypes[i]);
-        } else {
-          newData[i] = data[orderOfData[i]];
-        }
-      }
-      if (newData.length > noDictionaryMapping.length) {
-        for (int i = noDictionaryMapping.length; i < newData.length; i++) {
-          newData[i] = data[orderOfData[i]];
-        }
-      }
-      //      System.out.println(Arrays.toString(data));
-      return newData;
-    }
-
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/ec33c112/processing/src/main/java/org/apache/carbondata/processing/loading/steps/InputProcessorStepWithNoConverterImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/steps/InputProcessorStepWithNoConverterImpl.java
b/processing/src/main/java/org/apache/carbondata/processing/loading/steps/InputProcessorStepWithNoConverterImpl.java
new file mode 100644
index 0000000..8c24b7f
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/loading/steps/InputProcessorStepWithNoConverterImpl.java
@@ -0,0 +1,306 @@
+/*
+ * 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.steps;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.carbondata.common.CarbonIterator;
+import org.apache.carbondata.core.datastore.row.CarbonRow;
+import org.apache.carbondata.core.metadata.datatype.DataType;
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
+import org.apache.carbondata.core.metadata.encoder.Encoding;
+import org.apache.carbondata.core.util.CarbonProperties;
+import org.apache.carbondata.core.util.DataTypeUtil;
+import org.apache.carbondata.processing.datatypes.GenericDataType;
+import org.apache.carbondata.processing.loading.AbstractDataLoadProcessorStep;
+import org.apache.carbondata.processing.loading.CarbonDataLoadConfiguration;
+import org.apache.carbondata.processing.loading.DataField;
+import org.apache.carbondata.processing.loading.constants.DataLoadProcessorConstants;
+import org.apache.carbondata.processing.loading.converter.impl.FieldEncoderFactory;
+import org.apache.carbondata.processing.loading.converter.impl.RowConverterImpl;
+import org.apache.carbondata.processing.loading.exception.CarbonDataLoadingException;
+import org.apache.carbondata.processing.loading.row.CarbonRowBatch;
+import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
+
+/**
+ * It reads data from record reader and sends data to next step.
+ */
+public class InputProcessorStepWithNoConverterImpl extends AbstractDataLoadProcessorStep
{
+
+  private CarbonIterator<Object[]>[] inputIterators;
+
+  private boolean[] noDictionaryMapping;
+
+  private DataType[] dataTypes;
+
+  private int[] orderOfData;
+
+  private Map<Integer, GenericDataType> dataFieldsWithComplexDataType;
+
+  public InputProcessorStepWithNoConverterImpl(CarbonDataLoadConfiguration configuration,
+      CarbonIterator<Object[]>[] inputIterators) {
+    super(configuration, null);
+    this.inputIterators = inputIterators;
+  }
+
+  @Override public DataField[] getOutput() {
+    return configuration.getDataFields();
+  }
+
+  @Override public void initialize() throws IOException {
+    super.initialize();
+    // if logger is enabled then raw data will be required.
+    RowConverterImpl rowConverter =
+        new RowConverterImpl(configuration.getDataFields(), configuration, null);
+    rowConverter.initialize();
+    configuration.setCardinalityFinder(rowConverter);
+    noDictionaryMapping =
+        CarbonDataProcessorUtil.getNoDictionaryMapping(configuration.getDataFields());
+
+    dataFieldsWithComplexDataType = new HashMap<>();
+    convertComplexDataType(dataFieldsWithComplexDataType);
+
+    dataTypes = new DataType[configuration.getDataFields().length];
+    for (int i = 0; i < dataTypes.length; i++) {
+      if (configuration.getDataFields()[i].getColumn().hasEncoding(Encoding.DICTIONARY))
{
+        dataTypes[i] = DataTypes.INT;
+      } else {
+        dataTypes[i] = configuration.getDataFields()[i].getColumn().getDataType();
+      }
+    }
+    orderOfData = arrangeData(configuration.getDataFields(), configuration.getHeader());
+  }
+
+  private void convertComplexDataType(Map<Integer, GenericDataType> dataFieldsWithComplexDataType)
{
+    DataField[] srcDataField = configuration.getDataFields();
+    FieldEncoderFactory fieldConverterFactory = FieldEncoderFactory.getInstance();
+    String nullFormat =
+        configuration.getDataLoadProperty(DataLoadProcessorConstants.SERIALIZATION_NULL_FORMAT)
+            .toString();
+    boolean isEmptyBadRecord = Boolean.parseBoolean(
+        configuration.getDataLoadProperty(DataLoadProcessorConstants.IS_EMPTY_DATA_BAD_RECORD)
+            .toString());
+    for (int i = 0; i < srcDataField.length; i++) {
+      if (srcDataField[i].getColumn().isComplex()) {
+        // create a ComplexDataType
+        dataFieldsWithComplexDataType.put(srcDataField[i].getColumn().getOrdinal(),
+            fieldConverterFactory
+                .createComplexDataType(srcDataField[i], null, configuration.getTableIdentifier(),
+                    null, false, null, i, nullFormat, isEmptyBadRecord));
+      }
+    }
+  }
+
+  private int[] arrangeData(DataField[] dataFields, String[] header) {
+    int[] data = new int[dataFields.length];
+    for (int i = 0; i < dataFields.length; i++) {
+      for (int j = 0; j < header.length; j++) {
+        if (dataFields[i].getColumn().getColName().equalsIgnoreCase(header[j])) {
+          data[i] = j;
+          break;
+        }
+      }
+    }
+    return data;
+  }
+
+  @Override public Iterator<CarbonRowBatch>[] execute() {
+    int batchSize = CarbonProperties.getInstance().getBatchSize();
+    List<CarbonIterator<Object[]>>[] readerIterators = partitionInputReaderIterators();
+    Iterator<CarbonRowBatch>[] outIterators = new Iterator[readerIterators.length];
+    for (int i = 0; i < outIterators.length; i++) {
+      outIterators[i] =
+          new InputProcessorIterator(readerIterators[i], batchSize, configuration.isPreFetch(),
+              rowCounter, orderOfData, noDictionaryMapping, dataTypes,
+              configuration.getDataFields(), dataFieldsWithComplexDataType);
+    }
+    return outIterators;
+  }
+
+  /**
+   * Partition input iterators equally as per the number of threads.
+   *
+   * @return
+   */
+  private List<CarbonIterator<Object[]>>[] partitionInputReaderIterators() {
+    // Get the number of cores configured in property.
+    int numberOfCores = CarbonProperties.getInstance().getNumberOfCores();
+    // Get the minimum of number of cores and iterators size to get the number of parallel
threads
+    // to be launched.
+    int parallelThreadNumber = Math.min(inputIterators.length, numberOfCores);
+
+    List<CarbonIterator<Object[]>>[] iterators = new List[parallelThreadNumber];
+    for (int i = 0; i < parallelThreadNumber; i++) {
+      iterators[i] = new ArrayList<>();
+    }
+    // Equally partition the iterators as per number of threads
+    for (int i = 0; i < inputIterators.length; i++) {
+      iterators[i % parallelThreadNumber].add(inputIterators[i]);
+    }
+    return iterators;
+  }
+
+  @Override protected CarbonRow processRow(CarbonRow row) {
+    return null;
+  }
+
+  @Override public void close() {
+    if (!closed) {
+      super.close();
+      for (CarbonIterator inputIterator : inputIterators) {
+        inputIterator.close();
+      }
+    }
+  }
+
+  @Override protected String getStepName() {
+    return "Input Processor";
+  }
+
+  /**
+   * 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<CarbonIterator<Object[]>> inputIterators;
+
+    private CarbonIterator<Object[]> currentIterator;
+
+    private int counter;
+
+    private int batchSize;
+
+    private boolean nextBatch;
+
+    private boolean firstTime;
+
+    private AtomicLong rowCounter;
+
+    private boolean[] noDictionaryMapping;
+
+    private DataType[] dataTypes;
+
+    private DataField[] dataFields;
+
+    private int[] orderOfData;
+
+    private Map<Integer, GenericDataType> dataFieldsWithComplexDataType;
+
+    public InputProcessorIterator(List<CarbonIterator<Object[]>> inputIterators,
int batchSize,
+        boolean preFetch, AtomicLong rowCounter, int[] orderOfData, boolean[] noDictionaryMapping,
+        DataType[] dataTypes, DataField[] dataFields,
+        Map<Integer, GenericDataType> dataFieldsWithComplexDataType) {
+      this.inputIterators = inputIterators;
+      this.batchSize = batchSize;
+      this.counter = 0;
+      // Get the first iterator from the list.
+      currentIterator = inputIterators.get(counter++);
+      this.rowCounter = rowCounter;
+      this.nextBatch = false;
+      this.firstTime = true;
+      this.noDictionaryMapping = noDictionaryMapping;
+      this.dataTypes = dataTypes;
+      this.dataFields = dataFields;
+      this.orderOfData = orderOfData;
+      this.dataFieldsWithComplexDataType = dataFieldsWithComplexDataType;
+    }
+
+    @Override public boolean hasNext() {
+      return nextBatch || internalHasNext();
+    }
+
+    private boolean internalHasNext() {
+      if (firstTime) {
+        firstTime = false;
+        currentIterator.initialize();
+      }
+      boolean hasNext = currentIterator.hasNext();
+      // If iterator is finished then check for next iterator.
+      if (!hasNext) {
+        currentIterator.close();
+        // Check next iterator is available in the list.
+        if (counter < inputIterators.size()) {
+          // 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() {
+      return getBatch();
+    }
+
+    private CarbonRowBatch getBatch() {
+      // Create batch and fill it.
+      CarbonRowBatch carbonRowBatch = new CarbonRowBatch(batchSize);
+      int count = 0;
+      while (internalHasNext() && count < batchSize) {
+        carbonRowBatch.addRow(
+            new CarbonRow(convertToNoDictionaryToBytes(currentIterator.next(), dataFields)));
+        count++;
+      }
+      rowCounter.getAndAdd(carbonRowBatch.getSize());
+      return carbonRowBatch;
+    }
+
+    private Object[] convertToNoDictionaryToBytes(Object[] data, DataField[] dataFields)
{
+      Object[] newData = new Object[data.length];
+      for (int i = 0; i < data.length; i++) {
+        if (i < noDictionaryMapping.length && noDictionaryMapping[i]) {
+          newData[i] = DataTypeUtil
+              .getBytesDataDataTypeForNoDictionaryColumn(data[orderOfData[i]], dataTypes[i]);
+        } else {
+          // if this is a complex column then recursively comver the data into Byte Array.
+          if (dataTypes[i].isComplexType()) {
+            ByteArrayOutputStream byteArray = new ByteArrayOutputStream();
+            DataOutputStream dataOutputStream = new DataOutputStream(byteArray);
+            try {
+              GenericDataType complextType =
+                  dataFieldsWithComplexDataType.get(dataFields[i].getColumn().getOrdinal());
+
+              complextType.writeByteArray(data[orderOfData[i]], dataOutputStream);
+
+              dataOutputStream.close();
+              newData[i] = byteArray.toByteArray();
+            } catch (Exception e) {
+              throw new CarbonDataLoadingException("Loading Exception", e);
+            }
+          } else {
+            newData[i] = data[orderOfData[i]];
+          }
+        }
+      }
+      // System.out.println(Arrays.toString(data));
+      return newData;
+    }
+
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/ec33c112/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
index f47853e..f921fd5 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
@@ -240,6 +240,35 @@ public final class CarbonDataProcessorUtil {
         .toPrimitive(noDictionaryMapping.toArray(new Boolean[noDictionaryMapping.size()]));
   }
 
+  public static void getComplexNoDictionaryMapping(DataField[] dataFields,
+      List<Integer> complexNoDictionary) {
+
+    // save the Ordinal Number in the List.
+    for (DataField field : dataFields) {
+      if (field.getColumn().isComplex()) {
+        // get the childs.
+        getComplexNoDictionaryMapping(
+            ((CarbonDimension) field.getColumn()).getListOfChildDimensions(), complexNoDictionary);
+      }
+    }
+  }
+
+  public static void getComplexNoDictionaryMapping(List<CarbonDimension> carbonDimensions,
+      List<Integer> complexNoDictionary) {
+    for (CarbonDimension carbonDimension : carbonDimensions) {
+      if (carbonDimension.isComplex()) {
+        getComplexNoDictionaryMapping(carbonDimension.getListOfChildDimensions(),
+            complexNoDictionary);
+      } else {
+        // This is primitive type. Check the encoding for NoDictionary.
+        if (!carbonDimension.hasEncoding(Encoding.DICTIONARY)) {
+          complexNoDictionary.add(carbonDimension.getOrdinal());
+        }
+      }
+    }
+  }
+
+
   /**
    * Preparing the boolean [] to map whether the dimension use inverted index or not.
    */

http://git-wip-us.apache.org/repos/asf/carbondata/blob/ec33c112/store/sdk/src/main/java/org/apache/carbondata/sdk/file/AvroCarbonWriter.java
----------------------------------------------------------------------
diff --git a/store/sdk/src/main/java/org/apache/carbondata/sdk/file/AvroCarbonWriter.java
b/store/sdk/src/main/java/org/apache/carbondata/sdk/file/AvroCarbonWriter.java
index bc2e9db..946040f 100644
--- a/store/sdk/src/main/java/org/apache/carbondata/sdk/file/AvroCarbonWriter.java
+++ b/store/sdk/src/main/java/org/apache/carbondata/sdk/file/AvroCarbonWriter.java
@@ -26,6 +26,8 @@ import java.util.UUID;
 import org.apache.carbondata.common.annotations.InterfaceAudience;
 import org.apache.carbondata.hadoop.api.CarbonTableOutputFormat;
 import org.apache.carbondata.hadoop.internal.ObjectArrayWritable;
+import org.apache.carbondata.processing.loading.complexobjects.ArrayObject;
+import org.apache.carbondata.processing.loading.complexobjects.StructObject;
 import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
 
 import org.apache.avro.Schema;
@@ -70,15 +72,15 @@ class AvroCarbonWriter extends CarbonWriter {
       avroSchema = avroRecord.getSchema();
     }
     List<Schema.Field> fields = avroSchema.getFields();
-    Object [] csvField = new Object[fields.size()];
+    Object[] csvField = new Object[fields.size()];
     for (int i = 0; i < fields.size(); i++) {
-      csvField[i] = avroFieldToObject(fields.get(i), avroRecord.get(i), 0);
+      csvField[i] = avroFieldToObject(fields.get(i), avroRecord.get(i));
     }
     return csvField;
   }
 
-  private String avroFieldToObject(Schema.Field avroField, Object fieldValue, int delimiterLevel)
{
-    StringBuilder out = new StringBuilder();
+  private Object avroFieldToObject(Schema.Field avroField, Object fieldValue) {
+    Object out = new Object();
     Schema.Type type = avroField.schema().getType();
     switch (type) {
       case BOOLEAN:
@@ -86,55 +88,39 @@ class AvroCarbonWriter extends CarbonWriter {
       case LONG:
       case DOUBLE:
       case STRING:
+        out = fieldValue;
+        break;
       case FLOAT:
-        out.append(fieldValue.toString());
+        Float f = (Float) fieldValue;
+        out = f.doubleValue();
         break;
       case RECORD:
         List<Schema.Field> fields = avroField.schema().getFields();
-        String delimiter = null;
-        delimiterLevel ++;
+
+        Object[] structChildObjects = new Object[fields.size()];
         for (int i = 0; i < fields.size(); i++) {
-          if (delimiterLevel == 1) {
-            delimiter = "$";
-          } else if (delimiterLevel > 1) {
-            delimiter = ":";
-          }
-          if (i != (fields.size() - 1)) {
-            out.append(avroFieldToObject(fields.get(i), ((GenericData.Record) fieldValue).get(i),
-                delimiterLevel)).append(delimiter);
-          } else {
-            out.append(avroFieldToObject(fields.get(i), ((GenericData.Record) fieldValue).get(i),
-                delimiterLevel));
-          }
+          structChildObjects[i] =
+              avroFieldToObject(fields.get(i), ((GenericData.Record) fieldValue).get(i));
         }
+        StructObject structObject = new StructObject(structChildObjects);
+        out = structObject;
         break;
       case ARRAY:
         int size = ((ArrayList) fieldValue).size();
-        String delimiterArray = null;
-        delimiterLevel ++;
-        if (delimiterLevel == 1) {
-          delimiterArray = "$";
-        } else if (delimiterLevel > 1) {
-          delimiterArray = ":";
-        }
-
+        Object[] arrayChildObjects = new Object[size];
         for (int i = 0; i < size; i++) {
-          if (i != size - 1) {
-            out.append(avroFieldToObject(
-                new Schema.Field(avroField.name(), avroField.schema().getElementType(), null,
true),
-                ((ArrayList) fieldValue).get(i), delimiterLevel)).append(delimiterArray);
-          } else {
-            out.append(avroFieldToObject(
-                new Schema.Field(avroField.name(), avroField.schema().getElementType(), null,
true),
-                ((ArrayList) fieldValue).get(i), delimiterLevel));
-          }
+          arrayChildObjects[i] = (avroFieldToObject(
+              new Schema.Field(avroField.name(), avroField.schema().getElementType(), null,
true),
+              ((ArrayList) fieldValue).get(i)));
         }
+        ArrayObject arrayObject = new ArrayObject(arrayChildObjects);
+        out = arrayObject;
         break;
 
       default:
         throw new UnsupportedOperationException();
     }
-    return out.toString();
+    return out;
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/carbondata/blob/ec33c112/store/sdk/src/main/java/org/apache/carbondata/sdk/file/CarbonWriterBuilder.java
----------------------------------------------------------------------
diff --git a/store/sdk/src/main/java/org/apache/carbondata/sdk/file/CarbonWriterBuilder.java
b/store/sdk/src/main/java/org/apache/carbondata/sdk/file/CarbonWriterBuilder.java
index 397f151..76a46d0 100644
--- a/store/sdk/src/main/java/org/apache/carbondata/sdk/file/CarbonWriterBuilder.java
+++ b/store/sdk/src/main/java/org/apache/carbondata/sdk/file/CarbonWriterBuilder.java
@@ -330,6 +330,12 @@ public class CarbonWriterBuilder {
     Objects.requireNonNull(schema, "schema should not be null");
     Objects.requireNonNull(path, "path should not be null");
     CarbonLoadModel loadModel = createLoadModel();
+
+    // AVRO records are pushed to Carbon as Object not as Strings. This was done in order
to
+    // handle multi level complex type support. As there are no conversion converter step
is
+    // removed from the load. LoadWithoutConverter flag is going to point to the Loader Builder
+    // which will skip Conversion Step.
+    loadModel.setLoadWithoutCoverterStep(true);
     return new AvroCarbonWriter(loadModel);
   }
 


Mime
View raw message