carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jack...@apache.org
Subject [26/50] [abbrv] carbondata git commit: [CARBONDATA-2704] Index file size in describe formatted command is not updated correctly with the segment file
Date Wed, 18 Jul 2018 02:19:59 GMT
[CARBONDATA-2704] Index file size in describe formatted command is not updated correctly with
the segment file

Problem:
Describe formatted command is not showing correct index files size after index files merge.
Solution:
Segment file should be updated with the actual index files size of that segment after index
files merge.

This closes #2462


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

Branch: refs/heads/carbonstore
Commit: eb604fdb73983dfe9396d488a51907d90ed51d3e
Parents: cdee81d
Author: dhatchayani <dhatcha.official@gmail.com>
Authored: Mon Jul 9 11:19:51 2018 +0530
Committer: manishgupta88 <tomanishgupta18@gmail.com>
Committed: Sun Jul 15 20:34:32 2018 +0530

----------------------------------------------------------------------
 .../core/metadata/SegmentFileStore.java         |  4 +-
 .../apache/carbondata/core/util/CarbonUtil.java | 48 ++++++++++++-------
 .../core/writer/CarbonIndexFileMergeWriter.java | 17 +++----
 .../CarbonIndexFileMergeTestCase.scala          | 50 ++++++++++++++++++++
 .../spark/rdd/CarbonDataRDDFactory.scala        |  4 +-
 5 files changed, 96 insertions(+), 27 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/carbondata/blob/eb604fdb/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
b/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
index 3d3b245..ce79e65 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
@@ -281,7 +281,7 @@ public class SegmentFileStore {
    * @throws IOException
    */
   public static boolean updateSegmentFile(String tablePath, String segmentId, String segmentFile,
-      String tableId) throws IOException {
+      String tableId, SegmentFileStore segmentFileStore) throws IOException {
     boolean status = false;
     String tableStatusPath = CarbonTablePath.getTableStatusFilePath(tablePath);
     if (!FileFactory.isFileExist(tableStatusPath)) {
@@ -308,6 +308,8 @@ public class SegmentFileStore {
           // if the segments is in the list of marked for delete then update the status.
           if (segmentId.equals(detail.getLoadName())) {
             detail.setSegmentFile(segmentFile);
+            detail.setIndexSize(String.valueOf(CarbonUtil
+                .getCarbonIndexSize(segmentFileStore, segmentFileStore.getLocationMap())));
             break;
           }
         }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/eb604fdb/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
index e87e52c..9796696 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
@@ -2647,23 +2647,7 @@ public final class CarbonUtil {
       fileStore.readIndexFiles();
       Map<String, List<String>> indexFilesMap = fileStore.getIndexFilesMap();
       // get the size of carbonindex file
-      for (Map.Entry<String, SegmentFileStore.FolderDetails> entry : locationMap.entrySet())
{
-        SegmentFileStore.FolderDetails folderDetails = entry.getValue();
-        Set<String> carbonindexFiles = folderDetails.getFiles();
-        String mergeFileName = folderDetails.getMergeFileName();
-        if (null != mergeFileName) {
-          String mergeIndexPath =
-              fileStore.getTablePath() + entry.getKey() + CarbonCommonConstants.FILE_SEPARATOR
-                  + mergeFileName;
-          carbonIndexSize += FileFactory.getCarbonFile(mergeIndexPath).getSize();
-        }
-        for (String indexFile : carbonindexFiles) {
-          String indexPath =
-              fileStore.getTablePath() + entry.getKey() + CarbonCommonConstants.FILE_SEPARATOR
-                  + indexFile;
-          carbonIndexSize += FileFactory.getCarbonFile(indexPath).getSize();
-        }
-      }
+      carbonIndexSize = getCarbonIndexSize(fileStore, locationMap);
       for (Map.Entry<String, List<String>> entry : indexFilesMap.entrySet())
{
         // get the size of carbondata files
         for (String blockFile : entry.getValue()) {
@@ -2676,6 +2660,36 @@ public final class CarbonUtil {
     return dataAndIndexSize;
   }
 
+  /**
+   * Calcuate the index files size of the segment
+   *
+   * @param fileStore
+   * @param locationMap
+   * @return
+   */
+  public static long getCarbonIndexSize(SegmentFileStore fileStore,
+      Map<String, SegmentFileStore.FolderDetails> locationMap) {
+    long carbonIndexSize = 0L;
+    for (Map.Entry<String, SegmentFileStore.FolderDetails> entry : locationMap.entrySet())
{
+      SegmentFileStore.FolderDetails folderDetails = entry.getValue();
+      Set<String> carbonindexFiles = folderDetails.getFiles();
+      String mergeFileName = folderDetails.getMergeFileName();
+      if (null != mergeFileName) {
+        String mergeIndexPath =
+            fileStore.getTablePath() + entry.getKey() + CarbonCommonConstants.FILE_SEPARATOR
+                + mergeFileName;
+        carbonIndexSize += FileFactory.getCarbonFile(mergeIndexPath).getSize();
+      }
+      for (String indexFile : carbonindexFiles) {
+        String indexPath =
+            fileStore.getTablePath() + entry.getKey() + CarbonCommonConstants.FILE_SEPARATOR
+                + indexFile;
+        carbonIndexSize += FileFactory.getCarbonFile(indexPath).getSize();
+      }
+    }
+    return carbonIndexSize;
+  }
+
   // Get the total size of carbon data and the total size of carbon index
   public static HashMap<String, Long> getDataSizeAndIndexSize(String tablePath,
       Segment segment) throws IOException {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/eb604fdb/core/src/main/java/org/apache/carbondata/core/writer/CarbonIndexFileMergeWriter.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/writer/CarbonIndexFileMergeWriter.java
b/core/src/main/java/org/apache/carbondata/core/writer/CarbonIndexFileMergeWriter.java
index c293064..80a46cb 100644
--- a/core/src/main/java/org/apache/carbondata/core/writer/CarbonIndexFileMergeWriter.java
+++ b/core/src/main/java/org/apache/carbondata/core/writer/CarbonIndexFileMergeWriter.java
@@ -114,11 +114,11 @@ public class CarbonIndexFileMergeWriter {
   private String writeMergeIndexFileBasedOnSegmentFile(
       String segmentId,
       List<String> indexFileNamesTobeAdded,
-      SegmentFileStore sfs, CarbonFile[] indexFiles) throws IOException {
+      SegmentFileStore segmentFileStore, CarbonFile[] indexFiles) throws IOException {
     SegmentIndexFileStore fileStore = new SegmentIndexFileStore();
     fileStore
-        .readAllIIndexOfSegment(sfs.getSegmentFile(), sfs.getTablePath(), SegmentStatus.SUCCESS,
-            true);
+        .readAllIIndexOfSegment(segmentFileStore.getSegmentFile(), segmentFileStore.getTablePath(),
+            SegmentStatus.SUCCESS, true);
     Map<String, byte[]> indexMap = fileStore.getCarbonIndexMapWithFullPath();
     Map<String, Map<String, byte[]>> indexLocationMap = new HashMap<>();
     for (Map.Entry<String, byte[]> entry: indexMap.entrySet()) {
@@ -133,11 +133,12 @@ public class CarbonIndexFileMergeWriter {
     for (Map.Entry<String, Map<String, byte[]>> entry : indexLocationMap.entrySet())
{
       String mergeIndexFile =
           writeMergeIndexFile(indexFileNamesTobeAdded, entry.getKey(), entry.getValue(),
segmentId);
-      for (Map.Entry<String, SegmentFileStore.FolderDetails> segentry : sfs.getLocationMap()
-          .entrySet()) {
+      for (Map.Entry<String, SegmentFileStore.FolderDetails> segentry : segmentFileStore
+          .getLocationMap().entrySet()) {
         String location = segentry.getKey();
         if (segentry.getValue().isRelative()) {
-          location = sfs.getTablePath() + CarbonCommonConstants.FILE_SEPARATOR + location;
+          location =
+              segmentFileStore.getTablePath() + CarbonCommonConstants.FILE_SEPARATOR + location;
         }
         if (new Path(entry.getKey()).equals(new Path(location))) {
           segentry.getValue().setMergeFileName(mergeIndexFile);
@@ -153,9 +154,9 @@ public class CarbonIndexFileMergeWriter {
             + CarbonTablePath.SEGMENT_EXT;
     String path = CarbonTablePath.getSegmentFilesLocation(table.getTablePath())
         + CarbonCommonConstants.FILE_SEPARATOR + newSegmentFileName;
-    SegmentFileStore.writeSegmentFile(sfs.getSegmentFile(), path);
+    SegmentFileStore.writeSegmentFile(segmentFileStore.getSegmentFile(), path);
     SegmentFileStore.updateSegmentFile(table.getTablePath(), segmentId, newSegmentFileName,
-        table.getCarbonTableIdentifier().getTableId());
+        table.getCarbonTableIdentifier().getTableId(), segmentFileStore);
 
     for (CarbonFile file : indexFiles) {
       file.delete();

http://git-wip-us.apache.org/repos/asf/carbondata/blob/eb604fdb/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CarbonIndexFileMergeTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CarbonIndexFileMergeTestCase.scala
b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CarbonIndexFileMergeTestCase.scala
index b4937e6..51e46f7 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CarbonIndexFileMergeTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datacompaction/CarbonIndexFileMergeTestCase.scala
@@ -17,15 +17,20 @@
 
 package org.apache.carbondata.spark.testsuite.datacompaction
 
+import org.junit.Assert
+
 import scala.collection.JavaConverters._
 
 import org.apache.spark.sql.test.util.QueryTest
 import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
 
 import org.apache.carbondata.core.datamap.Segment
+import org.apache.carbondata.core.datastore.filesystem.{CarbonFile, CarbonFileFilter}
 import org.apache.carbondata.core.datastore.impl.FileFactory
 import org.apache.carbondata.core.indexstore.blockletindex.SegmentIndexFileStore
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable
 import org.apache.carbondata.core.metadata.{CarbonMetadata, SegmentFileStore}
+import org.apache.carbondata.core.statusmanager.SegmentStatusManager
 import org.apache.carbondata.core.util.path.CarbonTablePath
 import org.apache.carbondata.core.writer.CarbonIndexFileMergeWriter
 
@@ -193,6 +198,34 @@ class CarbonIndexFileMergeTestCase
     sql("select * from mitable").show()
   }
 
+  // CARBONDATA-2704, test the index file size after merge
+  test("Verify the size of the index file after merge") {
+    sql("DROP TABLE IF EXISTS fileSize")
+    sql(
+      """
+        | CREATE TABLE fileSize(id INT, name STRING, city STRING, age INT)
+        | STORED BY 'org.apache.carbondata.format'
+        | TBLPROPERTIES('SORT_COLUMNS'='city,name')
+      """.stripMargin)
+    sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE fileSize OPTIONS('header'='false')")
+    val table = CarbonMetadata.getInstance().getCarbonTable("default", "fileSize")
+    var loadMetadataDetails = SegmentStatusManager
+      .readTableStatusFile(CarbonTablePath.getTableStatusFilePath(table.getTablePath))
+    var segment0 = loadMetadataDetails.filter(x=> x.getLoadName.equalsIgnoreCase("0"))
+    Assert
+      .assertEquals(getIndexOrMergeIndexFileSize(table, "0", CarbonTablePath.INDEX_FILE_EXT),
+        segment0.head.getIndexSize.toLong)
+    new CarbonIndexFileMergeWriter(table)
+      .mergeCarbonIndexFilesOfSegment("0", table.getTablePath, false)
+    loadMetadataDetails = SegmentStatusManager
+      .readTableStatusFile(CarbonTablePath.getTableStatusFilePath(table.getTablePath))
+    segment0 = loadMetadataDetails.filter(x=> x.getLoadName.equalsIgnoreCase("0"))
+    Assert
+      .assertEquals(getIndexOrMergeIndexFileSize(table, "0", CarbonTablePath.MERGE_INDEX_FILE_EXT),
+        segment0.head.getIndexSize.toLong)
+    sql("DROP TABLE IF EXISTS fileSize")
+  }
+
   private def getIndexFileCount(tableName: String, segmentNo: String): Int = {
     val carbonTable = CarbonMetadata.getInstance().getCarbonTable(tableName)
     val segmentDir = CarbonTablePath.getSegmentPath(carbonTable.getTablePath, segmentNo)
@@ -222,4 +255,21 @@ class CarbonIndexFileMergeTestCase
     }
   }
 
+  private def getIndexOrMergeIndexFileSize(carbonTable: CarbonTable,
+      segmentId: String,
+      fileExtension: String): Long = {
+    var size = 0L;
+    val segmentPath = CarbonTablePath.getSegmentPath(carbonTable.getTablePath, segmentId)
+    val segmentFile = FileFactory.getCarbonFile(segmentPath)
+    val carbonFiles = segmentFile.listFiles(new CarbonFileFilter() {
+      override def accept(file: CarbonFile): Boolean = {
+        (file.getName.endsWith(fileExtension))
+      }
+    })
+    carbonFiles.toList.foreach(carbonFile => {
+      size += FileFactory.getCarbonFile(carbonFile.getPath).getSize
+    })
+    size
+  }
+
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/eb604fdb/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
index 543ba30..40d5c0d 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -522,7 +522,9 @@ object CarbonDataRDDFactory {
         carbonTable.getTablePath,
         carbonLoadModel.getSegmentId,
         segmentFileName,
-        carbonTable.getCarbonTableIdentifier.getTableId)
+        carbonTable.getCarbonTableIdentifier.getTableId,
+        new SegmentFileStore(carbonTable.getTablePath, segmentFileName))
+
       operationContext.setProperty(carbonTable.getTableUniqueName + "_Segment",
         carbonLoadModel.getSegmentId)
       val loadTablePreStatusUpdateEvent: LoadTablePreStatusUpdateEvent =


Mime
View raw message