You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ra...@apache.org on 2019/05/17 07:26:15 UTC
[carbondata] branch branch-1.5 updated (32b7471 -> 8a38edc)
This is an automated email from the ASF dual-hosted git repository.
ravipesala pushed a change to branch branch-1.5
in repository https://gitbox.apache.org/repos/asf/carbondata.git.
discard 32b7471 [maven-release-plugin] prepare for next development iteration
discard 7b880ba [maven-release-plugin] prepare release apache-carbondata-1.5.4-rc1
new 8a38edc [CARBONDATA-3386] Concurrent Merge index and query is failing
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (32b7471)
\
N -- N -- N refs/heads/branch-1.5 (8a38edc)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
assembly/pom.xml | 2 +-
common/pom.xml | 2 +-
core/pom.xml | 2 +-
datamap/bloom/pom.xml | 2 +-
datamap/examples/pom.xml | 2 +-
datamap/lucene/pom.xml | 2 +-
datamap/mv/core/pom.xml | 2 +-
datamap/mv/plan/pom.xml | 2 +-
examples/spark2/pom.xml | 2 +-
format/pom.xml | 2 +-
hadoop/pom.xml | 2 +-
integration/hive/pom.xml | 2 +-
integration/presto/pom.xml | 2 +-
integration/spark-common-test/pom.xml | 2 +-
.../createTable/TestCreateTableIfNotExists.scala | 1 +
.../CarbonIndexFileMergeTestCase.scala | 4 +-
integration/spark-common/pom.xml | 2 +-
integration/spark-datasource/pom.xml | 2 +-
integration/spark2/pom.xml | 2 +-
.../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 +++++++++++++++++++++
pom.xml | 2 +-
processing/pom.xml | 2 +-
store/sdk/pom.xml | 2 +-
streaming/pom.xml | 2 +-
tools/cli/pom.xml | 2 +-
29 files changed, 150 insertions(+), 92 deletions(-)
create mode 100644 integration/spark2/src/main/scala/org/apache/spark/util/MergeIndexUtil.scala
[carbondata] 01/01: [CARBONDATA-3386] Concurrent Merge index and
query is failing
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ravipesala pushed a commit to branch branch-1.5
in repository https://gitbox.apache.org/repos/asf/carbondata.git
commit 8a38edceb4246a9ae8627ad2e7ee0fbad77acf40
Author: dhatchayani <dh...@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)
+ }
+ }
+
+}