carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ravipes...@apache.org
Subject [carbondata] branch master updated: [CARBONDATA-3386] Concurrent Merge index and query is failing
Date Fri, 17 May 2019 07:19:32 GMT
This is an automated email from the ASF dual-hosted git repository.

ravipesala pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/carbondata.git


The following commit(s) were added to refs/heads/master by this push:
     new 894216e  [CARBONDATA-3386] Concurrent Merge index and query is failing
894216e is described below

commit 894216ebcf3de5a0df2f3c7c3e455d93530cc3a3
Author: dhatchayani <dhatcha.official@gmail.com>
AuthorDate: Wed May 15 14:48:20 2019 +0530

    [CARBONDATA-3386] Concurrent Merge index and query is failing
    
    Problem:
    Concurrent merge index and query is failing. Load is triggered on a table, at the end
of the load Merge index will be triggered. But this is triggered after the table status is
updated as SUCCESS/PARTIAL SUCCESS for that segments. So for the concurrent query, this segment
is available for query. Once the merge index is done, it deletes the index files, which are
still referred by the query, this leads to the query failure.
    
    Solution:
    Trigger merge index before table status updation for all the operations like LOAD, COMPACT.
    
    This closes #3221
---
 .../createTable/TestCreateTableIfNotExists.scala   |   1 +
 .../CarbonIndexFileMergeTestCase.scala             |   4 +-
 .../spark/rdd/CarbonTableCompactor.scala           |   5 +
 .../scala/org/apache/spark/sql/CarbonEnv.scala     |   5 +-
 .../spark/sql/events/MergeIndexEventListener.scala |  70 ++-----------
 .../command/table/CarbonCreateTableCommand.scala   |   5 +-
 .../org/apache/spark/util/MergeIndexUtil.scala     | 108 +++++++++++++++++++++
 7 files changed, 128 insertions(+), 70 deletions(-)

diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestCreateTableIfNotExists.scala
b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestCreateTableIfNotExists.scala
index dc54127..b3fa0eb 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestCreateTableIfNotExists.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestCreateTableIfNotExists.scala
@@ -78,6 +78,7 @@ class TestCreateTableIfNotExists extends QueryTest with BeforeAndAfterAll
{
           } catch {
             case exception: Exception =>
               result = exception.getMessage
+              exception.printStackTrace()
           }
           result
         }
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 173c14f..c9a7971 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
@@ -300,8 +300,8 @@ class CarbonIndexFileMergeTestCase
     assert(getIndexFileCount("default_nonindexmerge", "1") == 100)
     assert(getIndexFileCount("default_nonindexmerge", "2") == 100)
     assert(getIndexFileCount("default_nonindexmerge", "3") == 100)
-    assert(getIndexFileCount("default_nonindexmerge", "0.1") == 100)
-    assert(getIndexFileCount("default_nonindexmerge", "2.1") == 100)
+    assert(getIndexFileCount("default_nonindexmerge", "0.1") == 0)
+    assert(getIndexFileCount("default_nonindexmerge", "2.1") == 0)
     assert(getIndexFileCount("default_nonindexmerge", "0.2") == 0)
     checkAnswer(sql("""Select count(*) from nonindexmerge"""), rows)
   }
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
index cfb40ec..afe2927 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
@@ -26,6 +26,7 @@ import scala.collection.mutable
 
 import org.apache.spark.sql.SQLContext
 import org.apache.spark.sql.execution.command.{CarbonMergerMapping, CompactionCallableModel,
CompactionModel}
+import org.apache.spark.util.MergeIndexUtil
 
 import org.apache.carbondata.core.constants.CarbonCommonConstants
 import org.apache.carbondata.core.datamap.{DataMapStoreManager, Segment}
@@ -196,6 +197,10 @@ class CarbonTableCompactor(carbonLoadModel: CarbonLoadModel,
       val mergedLoadNumber = CarbonDataMergerUtil.getLoadNumberFromLoadName(mergedLoadName)
       var segmentFilesForIUDCompact = new util.ArrayList[Segment]()
       var segmentFileName: String = null
+      if (compactionType != CompactionType.IUD_DELETE_DELTA &&
+          compactionType != CompactionType.IUD_UPDDEL_DELTA) {
+        MergeIndexUtil.mergeIndexFilesOnCompaction(compactionCallableModel)
+      }
       if (carbonTable.isHivePartitionTable) {
         val readPath =
           CarbonTablePath.getSegmentFilesLocation(carbonLoadModel.getTablePath) +
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
index 7ca9945..149e45e 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
@@ -37,7 +37,7 @@ import org.apache.carbondata.core.metadata.schema.table.CarbonTable
 import org.apache.carbondata.core.util._
 import org.apache.carbondata.datamap.{TextMatchMaxDocUDF, TextMatchUDF}
 import org.apache.carbondata.events._
-import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadMetadataEvent, LoadTablePostExecutionEvent,
LoadTablePostStatusUpdateEvent, LoadTablePreExecutionEvent, LoadTablePreStatusUpdateEvent}
+import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadMetadataEvent, LoadTablePostStatusUpdateEvent,
LoadTablePreExecutionEvent, LoadTablePreStatusUpdateEvent}
 import org.apache.carbondata.spark.rdd.SparkReadSupport
 import org.apache.carbondata.spark.readsupport.SparkRowReadSupportImpl
 
@@ -182,8 +182,7 @@ object CarbonEnv {
       .addListener(classOf[AlterTableDropPartitionPostStatusEvent],
         AlterTableDropPartitionPostStatusListener)
       .addListener(classOf[AlterTableDropPartitionMetaEvent], AlterTableDropPartitionMetaListener)
-      .addListener(classOf[LoadTablePostExecutionEvent], new MergeIndexEventListener)
-      .addListener(classOf[AlterTableCompactionPostEvent], new MergeIndexEventListener)
+      .addListener(classOf[LoadTablePreStatusUpdateEvent], new MergeIndexEventListener)
       .addListener(classOf[AlterTableMergeIndexEvent], new MergeIndexEventListener)
       .addListener(classOf[BuildDataMapPostExecutionEvent], new MergeBloomIndexEventListener)
       .addListener(classOf[DropTableCacheEvent], DropCachePreAggEventListener)
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/events/MergeIndexEventListener.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/events/MergeIndexEventListener.scala
index 35b73d6..7844f28 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/events/MergeIndexEventListener.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/events/MergeIndexEventListener.scala
@@ -23,20 +23,17 @@ import scala.collection.JavaConverters._
 import scala.collection.mutable
 
 import org.apache.spark.internal.Logging
-import org.apache.spark.SparkContext
 import org.apache.spark.rdd.CarbonMergeFilesRDD
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.util.CarbonException
+import org.apache.spark.util.MergeIndexUtil
 
 import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.constants.CarbonCommonConstants
 import org.apache.carbondata.core.datamap.Segment
 import org.apache.carbondata.core.locks.{CarbonLockFactory, LockUsage}
-import org.apache.carbondata.core.metadata.SegmentFileStore
-import org.apache.carbondata.core.metadata.schema.table.CarbonTable
 import org.apache.carbondata.core.statusmanager.SegmentStatusManager
-import org.apache.carbondata.events.{AlterTableCompactionPostEvent, AlterTableMergeIndexEvent,
Event, OperationContext, OperationEventListener}
-import org.apache.carbondata.processing.loading.events.LoadEvents.LoadTablePostExecutionEvent
+import org.apache.carbondata.events._
+import org.apache.carbondata.processing.loading.events.LoadEvents.LoadTablePreStatusUpdateEvent
 import org.apache.carbondata.processing.merger.CarbonDataMergerUtil
 
 class MergeIndexEventListener extends OperationEventListener with Logging {
@@ -44,7 +41,7 @@ class MergeIndexEventListener extends OperationEventListener with Logging
{
 
   override def onEvent(event: Event, operationContext: OperationContext): Unit = {
     event match {
-      case preStatusUpdateEvent: LoadTablePostExecutionEvent =>
+      case preStatusUpdateEvent: LoadTablePreStatusUpdateEvent =>
         LOGGER.info("Load post status event-listener called for merge index")
         val loadModel = preStatusUpdateEvent.getCarbonLoadModel
         val carbonTable = loadModel.getCarbonDataLoadSchema.getCarbonTable
@@ -52,7 +49,7 @@ class MergeIndexEventListener extends OperationEventListener with Logging
{
         val sparkSession = SparkSession.getActiveSession.get
         if(!carbonTable.isStreamingSink) {
           if (null != compactedSegments && !compactedSegments.isEmpty) {
-            mergeIndexFilesForCompactedSegments(sparkSession,
+            MergeIndexUtil.mergeIndexFilesForCompactedSegments(sparkSession,
               carbonTable,
               compactedSegments)
           } else {
@@ -67,17 +64,9 @@ class MergeIndexEventListener extends OperationEventListener with Logging
{
               carbonTable.getTablePath,
               carbonTable, false)
             // clear Block dataMap Cache
-            clearBlockDataMapCache(carbonTable, Seq(loadModel.getSegmentId))
+            MergeIndexUtil.clearBlockDataMapCache(carbonTable, Seq(loadModel.getSegmentId))
           }
         }
-      case alterTableCompactionPostEvent: AlterTableCompactionPostEvent =>
-        LOGGER.info("Merge index for compaction called")
-        val carbonTable = alterTableCompactionPostEvent.carbonTable
-        val mergedLoads = alterTableCompactionPostEvent.compactedLoads
-        val sparkSession = alterTableCompactionPostEvent.sparkSession
-        if(!carbonTable.isStreamingSink) {
-          mergeIndexFilesForCompactedSegments(sparkSession, carbonTable, mergedLoads)
-        }
       case alterTableMergeIndexEvent: AlterTableMergeIndexEvent =>
         val carbonMainTable = alterTableMergeIndexEvent.carbonTable
         val sparkSession = alterTableMergeIndexEvent.sparkSession
@@ -123,7 +112,7 @@ class MergeIndexEventListener extends OperationEventListener with Logging
{
                 mergeIndexProperty = true,
                 readFileFooterFromCarbonDataFile = true)
               // clear Block dataMap Cache
-              clearBlockDataMapCache(carbonMainTable, validSegmentIds)
+              MergeIndexUtil.clearBlockDataMapCache(carbonMainTable, validSegmentIds)
               val requestMessage = "Compaction request completed for table " +
                 s"${ carbonMainTable.getDatabaseName }.${ carbonMainTable.getTableName }"
               LOGGER.info(requestMessage)
@@ -142,49 +131,4 @@ class MergeIndexEventListener extends OperationEventListener with Logging
{
     }
   }
 
-  def mergeIndexFilesForCompactedSegments(sparkSession: SparkSession,
-    carbonTable: CarbonTable,
-    mergedLoads: util.List[String]): Unit = {
-    // get only the valid segments of the table
-    val validSegments: mutable.Buffer[Segment] = CarbonDataMergerUtil.getValidSegmentList(
-      carbonTable.getAbsoluteTableIdentifier).asScala
-    val mergedSegmentIds = new util.ArrayList[String]()
-    mergedLoads.asScala.foreach(mergedLoad => {
-      val loadName = mergedLoad
-        .substring(mergedLoad.indexOf(CarbonCommonConstants.LOAD_FOLDER) +
-                   CarbonCommonConstants.LOAD_FOLDER.length)
-      mergedSegmentIds.add(loadName)
-    })
-    val loadFolderDetailsArray = SegmentStatusManager
-      .readLoadMetadata(carbonTable.getMetadataPath)
-    val segmentFileNameMap: java.util.Map[String, String] = new util.HashMap[String, String]()
-    loadFolderDetailsArray.foreach(loadMetadataDetails => {
-      segmentFileNameMap
-        .put(loadMetadataDetails.getLoadName, String.valueOf(loadMetadataDetails.getLoadStartTime))
-    })
-    // filter out only the valid segments from the list of compacted segments
-    // Example: say compacted segments list contains 0.1, 3.1, 6.1, 0.2.
-    // In this list 0.1, 3.1 and 6.1 are compacted to 0.2 in the level 2 compaction.
-    // So, it is enough to do merge index only for 0.2 as it is the only valid segment in
this list
-    val validMergedSegIds = validSegments
-      .filter { seg => mergedSegmentIds.contains(seg.getSegmentNo) }.map(_.getSegmentNo)
-    if (null != validMergedSegIds && validMergedSegIds.nonEmpty) {
-      CarbonMergeFilesRDD.mergeIndexFiles(sparkSession,
-          validMergedSegIds,
-          segmentFileNameMap,
-          carbonTable.getTablePath,
-          carbonTable,
-          false)
-      // clear Block dataMap Cache
-      clearBlockDataMapCache(carbonTable, validMergedSegIds)
-    }
-  }
-
-  private def clearBlockDataMapCache(carbonTable: CarbonTable, segmentIds: Seq[String]):
Unit = {
-    // clear driver Block dataMap cache for each segment
-    segmentIds.foreach { segmentId =>
-      SegmentFileStore.clearBlockDataMapCache(carbonTable, segmentId)
-    }
-  }
-
 }
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
index debb283..b5aa8f9 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
@@ -191,10 +191,11 @@ case class CarbonCreateTableCommand(
               if (ifNotExistsSet) {
                 LOGGER.error(e, e)
               } else {
+                LOGGER.error(e)
                 throw e
               }
-            }
-            else {
+            } else {
+              LOGGER.error(e)
               throw e
             }
 
diff --git a/integration/spark2/src/main/scala/org/apache/spark/util/MergeIndexUtil.scala
b/integration/spark2/src/main/scala/org/apache/spark/util/MergeIndexUtil.scala
new file mode 100644
index 0000000..f109b2e
--- /dev/null
+++ b/integration/spark2/src/main/scala/org/apache/spark/util/MergeIndexUtil.scala
@@ -0,0 +1,108 @@
+/*
+ * 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.spark.util
+
+import java.util
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+import org.apache.spark.rdd.CarbonMergeFilesRDD
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.execution.command.CompactionCallableModel
+
+import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.datamap.Segment
+import org.apache.carbondata.core.metadata.SegmentFileStore
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable
+import org.apache.carbondata.core.statusmanager.SegmentStatusManager
+import org.apache.carbondata.processing.loading.model.CarbonLoadModel
+import org.apache.carbondata.processing.merger.CarbonDataMergerUtil
+
+object MergeIndexUtil {
+
+  val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
+
+  def mergeIndexFilesOnCompaction(compactionCallableModel: CompactionCallableModel): Unit
= {
+    val carbonTable = compactionCallableModel.carbonTable
+    LOGGER.info(s"Merge index for compaction is called on table ${carbonTable.getTableUniqueName}")
+    val mergedLoads = compactionCallableModel.compactedSegments
+    val sparkSession = compactionCallableModel.sqlContext.sparkSession
+    if (!carbonTable.isStreamingSink) {
+      val mergedSegmentIds = new util.ArrayList[String]()
+      mergedLoads.asScala.foreach(mergedLoad => {
+        val loadName = mergedLoad
+          .substring(mergedLoad.indexOf(CarbonCommonConstants.LOAD_FOLDER) +
+                     CarbonCommonConstants.LOAD_FOLDER.length)
+        mergedSegmentIds.add(loadName)
+      })
+      CarbonMergeFilesRDD.mergeIndexFiles(sparkSession,
+        mergedSegmentIds.asScala,
+        new util.HashMap[String, String](),
+        carbonTable.getTablePath,
+        carbonTable, false)
+    }
+  }
+
+  def mergeIndexFilesForCompactedSegments(sparkSession: SparkSession,
+    carbonTable: CarbonTable,
+    mergedLoads: util.List[String]): Unit = {
+    // get only the valid segments of the table
+    val validSegments: mutable.Buffer[Segment] = CarbonDataMergerUtil.getValidSegmentList(
+      carbonTable.getAbsoluteTableIdentifier).asScala
+    val mergedSegmentIds = new util.ArrayList[String]()
+    mergedLoads.asScala.foreach(mergedLoad => {
+      val loadName = mergedLoad
+        .substring(mergedLoad.indexOf(CarbonCommonConstants.LOAD_FOLDER) +
+                   CarbonCommonConstants.LOAD_FOLDER.length)
+      mergedSegmentIds.add(loadName)
+    })
+    val loadFolderDetailsArray = SegmentStatusManager
+      .readLoadMetadata(carbonTable.getMetadataPath)
+    val segmentFileNameMap: java.util.Map[String, String] = new util.HashMap[String, String]()
+    loadFolderDetailsArray.foreach(loadMetadataDetails => {
+      segmentFileNameMap
+        .put(loadMetadataDetails.getLoadName, String.valueOf(loadMetadataDetails.getLoadStartTime))
+    })
+    // filter out only the valid segments from the list of compacted segments
+    // Example: say compacted segments list contains 0.1, 3.1, 6.1, 0.2.
+    // In this list 0.1, 3.1 and 6.1 are compacted to 0.2 in the level 2 compaction.
+    // So, it is enough to do merge index only for 0.2 as it is the only valid segment in
this list
+    val validMergedSegIds = validSegments
+      .filter { seg => mergedSegmentIds.contains(seg.getSegmentNo) }.map(_.getSegmentNo)
+    if (null != validMergedSegIds && validMergedSegIds.nonEmpty) {
+      CarbonMergeFilesRDD.mergeIndexFiles(sparkSession,
+        validMergedSegIds,
+        segmentFileNameMap,
+        carbonTable.getTablePath,
+        carbonTable,
+        false)
+      // clear Block dataMap Cache
+      clearBlockDataMapCache(carbonTable, validMergedSegIds)
+    }
+  }
+
+  def clearBlockDataMapCache(carbonTable: CarbonTable, segmentIds: Seq[String]): Unit = {
+    // clear driver Block dataMap cache for each segment
+    segmentIds.foreach { segmentId =>
+      SegmentFileStore.clearBlockDataMapCache(carbonTable, segmentId)
+    }
+  }
+
+}


Mime
View raw message