carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jack...@apache.org
Subject [1/3] incubator-carbondata git commit: fix InsertInto issue for spark2
Date Sun, 15 Jan 2017 12:07:19 GMT
Repository: incubator-carbondata
Updated Branches:
  refs/heads/master 392bc290e -> 35f74248f


http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/30ee42cb/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
index f020a1c..8d50324 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
@@ -26,7 +26,6 @@ import scala.collection.mutable.ArrayBuffer
 import scala.collection.JavaConverters._
 import scala.util.parsing.combinator.RegexParsers
 
-import org.apache.spark.internal.Logging
 import org.apache.spark.sql.{RuntimeConfig, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation, NoSuchTableException}
@@ -39,8 +38,9 @@ import org.apache.carbondata.common.logging.LogServiceFactory
 import org.apache.carbondata.core.carbon.CarbonTableIdentifier
 import org.apache.carbondata.core.carbon.metadata.CarbonMetadata
 import org.apache.carbondata.core.carbon.metadata.converter.ThriftWrapperSchemaConverterImpl
+import org.apache.carbondata.core.carbon.metadata.datatype.DataType.DECIMAL
 import org.apache.carbondata.core.carbon.metadata.schema.table.CarbonTable
-import org.apache.carbondata.core.carbon.metadata.schema.table.column.CarbonDimension
+import org.apache.carbondata.core.carbon.metadata.schema.table.column.{CarbonColumn, CarbonDimension}
 import org.apache.carbondata.core.carbon.path.{CarbonStorePath, CarbonTablePath}
 import org.apache.carbondata.core.carbon.querystatistics.{QueryStatistic, QueryStatisticsConstants}
 import org.apache.carbondata.core.constants.CarbonCommonConstants
@@ -782,7 +782,48 @@ case class CarbonRelation(
         nullable = true)())
   }
 
-  override val output = dimensionsAttr ++ measureAttr
+  override val output = {
+    val columns = tableMeta.carbonTable.getCreateOrderColumn(tableMeta.carbonTable.getFactTableName)
+      .asScala
+    // convert each column to Attribute
+    columns.filter(!_.isInvisible).map { column =>
+      if (column.isDimesion()) {
+        val output: DataType = column.getDataType.toString.toLowerCase match {
+          case "array" =>
+            CarbonMetastoreTypes.toDataType(s"array<${getArrayChildren(column.getColName)}>")
+          case "struct" =>
+            CarbonMetastoreTypes.toDataType(s"struct<${getStructChildren(column.getColName)}>")
+          case dType =>
+            val dataType = addDecimalScaleAndPrecision(column, dType)
+            CarbonMetastoreTypes.toDataType(dataType)
+        }
+        AttributeReference(column.getColName, output, nullable = true )(
+          qualifier = Option(tableName + "." + column.getColName))
+      } else {
+        val output = CarbonMetastoreTypes.toDataType {
+          column.getDataType.toString
+            .toLowerCase match {
+            case "int" => "long"
+            case "short" => "long"
+            case "decimal" => "decimal(" + column.getColumnSchema.getPrecision + "," +
column
+              .getColumnSchema.getScale + ")"
+            case others => others
+          }
+        }
+        AttributeReference(column.getColName, output, nullable = true)(
+          qualifier = Option(tableName + "." + column.getColName))
+      }
+    }
+  }
+
+  def addDecimalScaleAndPrecision(dimval: CarbonColumn, dataType: String): String = {
+    var dType = dataType
+    if (dimval.getDataType == DECIMAL) {
+      dType +=
+      "(" + dimval.getColumnSchema.getPrecision + "," + dimval.getColumnSchema.getScale +
")"
+    }
+    dType
+  }
 
   // TODO: Use data from the footers.
   override lazy val statistics = Statistics(sizeInBytes = this.sizeInBytes)
@@ -797,8 +838,7 @@ case class CarbonRelation(
 
   def addDecimalScaleAndPrecision(dimval: CarbonDimension, dataType: String): String = {
     var dType = dataType
-    if (dimval.getDataType
-      == org.apache.carbondata.core.carbon.metadata.datatype.DataType.DECIMAL) {
+    if (dimval.getDataType == DECIMAL) {
       dType +=
         "(" + dimval.getColumnSchema.getPrecision + "," + dimval.getColumnSchema.getScale
+ ")"
     }

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/30ee42cb/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala
index 52c0eb0..a4fdc2f 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala
@@ -163,12 +163,15 @@ class CarbonLateDecodeRule extends Rule[LogicalPlan] with PredicateHelper
{
             Sort(sort.order, sort.global, child)
           }
         case union: Union
-          if !(union.children.head.isInstanceOf[CarbonDictionaryTempDecoder] ||
-            union.children(1).isInstanceOf[CarbonDictionaryTempDecoder]) =>
+          if !union.children.exists(_.isInstanceOf[CarbonDictionaryTempDecoder]) =>
           val children = union.children.map { child =>
             val condAttrs = new util.HashSet[AttributeReferenceWrapper]
             child.output.foreach(attr =>
-              condAttrs.add(AttributeReferenceWrapper(aliasMap.getOrElse(attr, attr))))
+              if (isDictionaryEncoded(attr, attrMap, aliasMap)) {
+                condAttrs.add(AttributeReferenceWrapper(aliasMap.getOrElse(attr, attr)))
+              }
+            )
+
             if (hasCarbonRelation(child) && condAttrs.size() > 0 &&
               !child.isInstanceOf[CarbonDictionaryCatalystDecoder]) {
               CarbonDictionaryTempDecoder(condAttrs,

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/30ee42cb/processing/src/main/java/org/apache/carbondata/processing/csvreaderstep/CsvInput.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/csvreaderstep/CsvInput.java
b/processing/src/main/java/org/apache/carbondata/processing/csvreaderstep/CsvInput.java
index f47babc..cbae5a8 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/csvreaderstep/CsvInput.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/csvreaderstep/CsvInput.java
@@ -91,6 +91,8 @@ public class CsvInput extends BaseStep implements StepInterface {
    */
   private String rddIteratorKey = null;
 
+  private CarbonIterator<CarbonIterator<String[]>> rddIterator;
+
   public CsvInput(StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr,
       TransMeta transMeta, Trans trans) {
     super(stepMeta, stepDataInterface, copyNr, transMeta, trans);
@@ -194,28 +196,27 @@ public class CsvInput extends BaseStep implements StepInterface {
   }
 
   class RddScanCallable implements Callable<Void> {
-    List<CarbonIterator<String[]>> iterList;
-
-    RddScanCallable() {
-      this.iterList = new ArrayList<CarbonIterator<String[]>>(1000);
-    }
-
-    public void addJavaRddIterator(CarbonIterator<String[]> iter) {
-      this.iterList.add(iter);
-    }
-
-    @Override
-    public Void call() throws Exception {
-      StandardLogService.setThreadName(("PROCESS_DataFrame_PARTITIONS"),
-          Thread.currentThread().getName());
+    @Override public Void call() throws Exception {
+      StandardLogService
+          .setThreadName(("PROCESS_DataFrame_PARTITIONS"), Thread.currentThread().getName());
       try {
         String[] values = null;
-        for (CarbonIterator<String[]> iter: iterList) {
-          iter.initialize();
-          while (iter.hasNext()) {
-            values = iter.next();
-            synchronized (putRowLock) {
-              putRow(data.outputRowMeta, values);
+        boolean hasNext = true;
+        CarbonIterator<String[]> iter;
+        boolean isInitialized = false;
+        while (hasNext) {
+          // Inovke getRddIterator to get a RDD[Row] iterator of a partition.
+          // The RDD comes from the sub-query DataFrame in InsertInto statement.
+          iter = getRddIterator(isInitialized);
+          isInitialized = true;
+          if (iter == null) {
+            hasNext = false;
+          } else {
+            while (iter.hasNext()) {
+              values = iter.next();
+              synchronized (putRowLock) {
+                putRow(data.outputRowMeta, values);
+              }
             }
           }
         }
@@ -225,34 +226,34 @@ public class CsvInput extends BaseStep implements StepInterface {
       }
       return null;
     }
-  };
+  }
+
+  private synchronized CarbonIterator<String[]> getRddIterator(boolean isInitialized)
{
+    if (!isInitialized) {
+      rddIterator.initialize();
+    }
+    if (rddIterator.hasNext()) {
+      return rddIterator.next();
+    }
+    return null;
+  }
 
   private void scanRddIterator(int numberOfNodes) throws RuntimeException {
-    CarbonIterator<CarbonIterator<String[]>> iter = RddInputUtils.getAndRemove(rddIteratorKey);
-    if (iter != null) {
-      iter.initialize();
+    rddIterator = RddInputUtils.getAndRemove(rddIteratorKey);
+    if (rddIterator != null) {
       exec = Executors.newFixedThreadPool(numberOfNodes);
       List<Future<Void>> results = new ArrayList<Future<Void>>(numberOfNodes);
       RddScanCallable[] calls = new RddScanCallable[numberOfNodes];
       for (int i = 0; i < numberOfNodes; i++ ) {
         calls[i] = new RddScanCallable();
-      }
-      int index = 0 ;
-      while (iter.hasNext()) {
-        calls[index].addJavaRddIterator(iter.next());
-        index = index + 1;
-        if (index == numberOfNodes) {
-          index = 0;
-        }
-      }
-      for (RddScanCallable call: calls) {
-        results.add(exec.submit(call));
+        results.add(exec.submit(calls[i]));
       }
       try {
         for (Future<Void> futrue : results) {
           futrue.get();
         }
       } catch (InterruptedException | ExecutionException e) {
+        LOGGER.error(e, "Thread InterruptedException");
         throw new RuntimeException("Thread InterruptedException", e);
       } finally {
         exec.shutdownNow();
@@ -264,7 +265,7 @@ public class CsvInput extends BaseStep implements StepInterface {
     Iterator<String[]> iterator = RddInpututilsForUpdate.getAndRemove(rddIteratorKey);
     if (iterator != null) {
       try{
-        while(iterator.hasNext()){
+        while (iterator.hasNext()) {
           putRow(data.outputRowMeta, iterator.next());
         }
       } catch (KettleException e) {
@@ -430,4 +431,4 @@ public class CsvInput extends BaseStep implements StepInterface {
     return false;
   }
 
-}
\ No newline at end of file
+}


Mime
View raw message