[CARBONDATA-2885] Broadcast Issue and Small file distribution Issue

Issue :- In External Table Carbon Relation sizeInByte is wrong (always 0) because of this Join Queries are identified for broadcast even Table actual size is > 10MB( default broadcast).This is making fail some of the join table ( table which should select sortmergeJoin but because of wrong calculation it gone for broadcast join) if Merge_small_file task distribution is enabled ,Join queries are failed (TPCH). carbon opens many carbon files but it not getting closed. Root Cause :- 1. Current relation size calculation is based on tablestatus file but since External Table does not have tablestatus file so always zero was returned. 2. if Merge_small_file task distribution is enabled carbon opens many carbon files but it not getting closed. Solution :- if Table is External Table then calculate size from TablePath . close the carbon files for scan is finished. This closes #2658
apache · Aug 27, 2018 · 1fb1f19 · 1fb1f19
1 parent f81543e
commit 1fb1f19
Show file tree

Hide file tree

Showing 2 changed files with 42 additions and 28 deletions.
diff --git a/...va/org/apache/carbondata/core/scan/result/iterator/AbstractDetailQueryResultIterator.java b/...va/org/apache/carbondata/core/scan/result/iterator/AbstractDetailQueryResultIterator.java
@@ -254,6 +254,11 @@ void updateDataBlockIterator() {
 
   private DataBlockIterator getDataBlockIterator() {
     if (blockExecutionInfos.size() > 0) {
+      try {
+        fileReader.finish();
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
       BlockExecutionInfo executionInfo = blockExecutionInfos.get(0);
       blockExecutionInfos.remove(executionInfo);
       return new DataBlockIterator(executionInfo, fileReader, batchSize, queryStatisticsModel,

diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonRelation.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonRelation.scala
@@ -156,39 +156,48 @@ case class CarbonRelation(
   private var sizeInBytesLocalValue = 0L
 
   def sizeInBytes: Long = {
-    val tableStatusNewLastUpdatedTime = SegmentStatusManager.getTableStatusLastModifiedTime(
-      carbonTable.getAbsoluteTableIdentifier)
-    if (tableStatusLastUpdateTime != tableStatusNewLastUpdatedTime) {
-      if (new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier)
-        .getValidAndInvalidSegments.getValidSegments.isEmpty) {
-        sizeInBytesLocalValue = 0L
-      } else {
-        val tablePath = carbonTable.getTablePath
-        val fileType = FileFactory.getFileType(tablePath)
-        if (FileFactory.isFileExist(tablePath, fileType)) {
-          // get the valid segments
-          val segments = new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier)
-            .getValidAndInvalidSegments.getValidSegments.asScala
-          var size = 0L
-          // for each segment calculate the size
-          segments.foreach {validSeg =>
-            // for older store
-            if (null != validSeg.getLoadMetadataDetails.getDataSize &&
-                null != validSeg.getLoadMetadataDetails.getIndexSize) {
-              size = size + validSeg.getLoadMetadataDetails.getDataSize.toLong +
-                     validSeg.getLoadMetadataDetails.getIndexSize.toLong
-            } else {
-              size = size + FileFactory.getDirectorySize(
-                CarbonTablePath.getSegmentPath(tablePath, validSeg.getSegmentNo))
+    if (carbonTable.isExternalTable) {
+      val tablePath = carbonTable.getTablePath
+      val fileType = FileFactory.getFileType(tablePath)
+      if (FileFactory.isFileExist(tablePath, fileType)) {
+        sizeInBytesLocalValue = FileFactory.getDirectorySize(tablePath)
+      }
+    } else {
+      val tableStatusNewLastUpdatedTime = SegmentStatusManager.getTableStatusLastModifiedTime(
+        carbonTable.getAbsoluteTableIdentifier)
+      if (tableStatusLastUpdateTime != tableStatusNewLastUpdatedTime) {
+        if (new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier)
+          .getValidAndInvalidSegments.getValidSegments.isEmpty) {
+          sizeInBytesLocalValue = 0L
+        } else {
+          val tablePath = carbonTable.getTablePath
+          val fileType = FileFactory.getFileType(tablePath)
+          if (FileFactory.isFileExist(tablePath, fileType)) {
+            // get the valid segments
+            val segments = new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier)
+              .getValidAndInvalidSegments.getValidSegments.asScala
+            var size = 0L
+            // for each segment calculate the size
+            segments.foreach { validSeg =>
+              // for older store
+              if (null != validSeg.getLoadMetadataDetails.getDataSize &&
+                  null != validSeg.getLoadMetadataDetails.getIndexSize) {
+                size = size + validSeg.getLoadMetadataDetails.getDataSize.toLong +
+                       validSeg.getLoadMetadataDetails.getIndexSize.toLong
+              } else {
+                size = size + FileFactory.getDirectorySize(
+                  CarbonTablePath.getSegmentPath(tablePath, validSeg.getSegmentNo))
+              }
             }
+            // update the new table status time
+            tableStatusLastUpdateTime = tableStatusNewLastUpdatedTime
+            // update the new size
+            sizeInBytesLocalValue = size
           }
-          // update the new table status time
-          tableStatusLastUpdateTime = tableStatusNewLastUpdatedTime
-          // update the new size
-          sizeInBytesLocalValue = size
         }
       }
     }
+
     sizeInBytesLocalValue
   }