This is an automated email from the ASF dual-hosted git repository.
liuzhi 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 1d316fc [CARBONDATA-3877] Reduce read tablestatus overhead during
inserting into partition table
1d316fc is described below
commit 1d316fc0ee3af9ecfda7208379051905b8460dde
Author: haomarch <marchp...@126.com>
AuthorDate: Sun Jun 28 19:21:19 2020 +0800
[CARBONDATA-3877] Reduce read tablestatus overhead during inserting into
partition table
Why is this PR needed?
Currently during inserting into a partition table, there are a lot of
tablestauts read operations, but when storing table status file in object
store, reading of table status file may fail (receive IOException or
JsonSyntaxException) when table status file is being modifying, which leading
to High failure rate when concurrent insert into a partition table.
What changes were proposed in this PR?
(1) Three codes was removed：calcute sizeinbytes, clean segments,
deleteLoadsAndUpdateMetadata
'calcute sizeinbytes' is useless during inserting into flow. 'clean
segments' and 'deleteLoadsAndUpdateMetadata' are supported by 'clean files'
command, which can be removed from inserting into flow.
(2) Reduce duplicate tablestatus operations and limit the conditions to get
tablestatus.
Does this PR introduce any user interface change?
No
Is any new testcase added?
No
This closes #3800
---
.../carbondata/core/metadata/SegmentFileStore.java | 17 +++++++++++++++++
.../hadoop/api/CarbonOutputCommitter.java | 11 +++++++----
.../org/apache/spark/rdd/CarbonMergeFilesRDD.scala | 7 ++++---
.../management/CarbonInsertIntoCommand.scala | 2 --
.../command/management/CommonLoadUtils.scala | 5 +----
.../datasources/SparkCarbonTableFormat.scala | 21 ++++++++++++++++-----
.../SILoadEventListenerForFailedSegments.scala | 3 ++-
.../loading/TableProcessingOperations.java | 3 +--
8 files changed, 48 insertions(+), 21 deletions(-)
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 335e0f5..2f274c3 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
@@ -1149,6 +1149,23 @@ public class SegmentFileStore {
* @throws IOException
*/
public static List<PartitionSpec> getPartitionSpecs(String segmentId, String
tablePath,
+ String segmentFilePath, String loadStartTime) throws IOException {
+ SegmentFileStore fileStore = new SegmentFileStore(tablePath,
segmentFilePath);
+ List<PartitionSpec> partitionSpecs = fileStore.getPartitionSpecs();
+ for (PartitionSpec spec : partitionSpecs) {
+ spec.setUuid(segmentId + "_" + loadStartTime);
+ }
+ return partitionSpecs;
+ }
+
+ /**
+ * Get the partition specs of the segment
+ * @param segmentId
+ * @param tablePath
+ * @return
+ * @throws IOException
+ */
+ public static List<PartitionSpec> getPartitionSpecs(String segmentId, String
tablePath,
LoadMetadataDetails[] details)
throws IOException {
LoadMetadataDetails segEntry = null;
diff --git
a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java
b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java
index 02c8d4c..4b8fc43 100644
---
a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java
+++
b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java
@@ -232,10 +232,13 @@ public class CarbonOutputCommitter extends
FileOutputCommitter {
String segmentsToBeDeleted =
context.getConfiguration().get(CarbonTableOutputFormat.SEGMENTS_TO_BE_DELETED,
"");
List<Segment> segmentDeleteList =
Segment.toSegmentList(segmentsToBeDeleted.split(","), null);
- Set<Segment> segmentSet = new HashSet<>(
- new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier(),
-
context.getConfiguration()).getValidAndInvalidSegments(carbonTable.isMV())
- .getValidSegments());
+ Set<Segment> segmentSet = new HashSet<>();
+ if (updateTime != null || uniqueId != null) {
+ segmentSet = new HashSet<>(
+ new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier(),
+
context.getConfiguration()).getValidAndInvalidSegments(carbonTable.isMV())
+ .getValidSegments());
+ }
if (updateTime != null) {
CarbonUpdateUtil.updateTableMetadataStatus(segmentSet, carbonTable,
updateTime, true,
segmentDeleteList);
diff --git
a/integration/spark/src/main/scala/org/apache/spark/rdd/CarbonMergeFilesRDD.scala
b/integration/spark/src/main/scala/org/apache/spark/rdd/CarbonMergeFilesRDD.scala
index ebac5e4..695ee27 100644
---
a/integration/spark/src/main/scala/org/apache/spark/rdd/CarbonMergeFilesRDD.scala
+++
b/integration/spark/src/main/scala/org/apache/spark/rdd/CarbonMergeFilesRDD.scala
@@ -229,14 +229,15 @@ class CarbonMergeFilesRDD(
override def internalGetPartitions: Array[Partition] = {
if (isHivePartitionedTable) {
- val metadataDetails = SegmentStatusManager
-
.readLoadMetadata(CarbonTablePath.getMetadataPath(carbonTable.getTablePath))
// in case of partition table make rdd partitions per partition of the
carbon table
val partitionPaths: java.util.Map[String, java.util.List[String]] = new
java.util.HashMap()
if (partitionInfo == null || partitionInfo.isEmpty) {
segments.foreach(segment => {
+ val loadStartTime = segmentFileNameToSegmentIdMap.get(segment)
+ val segmentFileName = SegmentFileStore.genSegmentFileName(
+ segment, loadStartTime) + CarbonTablePath.SEGMENT_EXT
val partitionSpecs = SegmentFileStore
- .getPartitionSpecs(segment, carbonTable.getTablePath,
metadataDetails)
+ .getPartitionSpecs(segment, carbonTable.getTablePath,
segmentFileName, loadStartTime)
.asScala.map(_.getLocation.toString)
partitionPaths.put(segment, partitionSpecs.asJava)
})
diff --git
a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertIntoCommand.scala
b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertIntoCommand.scala
index 8dfad76..8c14917 100644
---
a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertIntoCommand.scala
+++
b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertIntoCommand.scala
@@ -202,8 +202,6 @@ case class CarbonInsertIntoCommand(databaseNameOp:
Option[String],
updateModel = None,
operationContext = operationContext)
- // Clean up the old invalid segment data before creating a new entry for
new load.
- SegmentStatusManager.deleteLoadsAndUpdateMetadata(table, false,
currPartitions)
// add the start entry for the new load in the table status file
if (!table.isHivePartitionTable) {
CarbonLoaderUtil.readAndUpdateLoadProgressInTableMeta(
diff --git
a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CommonLoadUtils.scala
b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CommonLoadUtils.scala
index 20d29d8..71649b8 100644
---
a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CommonLoadUtils.scala
+++
b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CommonLoadUtils.scala
@@ -126,7 +126,6 @@ object CommonLoadUtils {
TableIdentifier(tableName, databaseNameOp))).collect {
case l: LogicalRelation => l
}.head
- sizeInBytes = logicalPartitionRelation.relation.sizeInBytes
finalPartition = getCompletePartitionValues(partition, table)
}
(sizeInBytes, table, dbName, logicalPartitionRelation, finalPartition)
@@ -843,8 +842,6 @@ object CommonLoadUtils {
def loadDataWithPartition(loadParams: CarbonLoadParams): Seq[Row] = {
val table =
loadParams.carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
val catalogTable: CatalogTable =
loadParams.logicalPartitionRelation.catalogTable.get
- // Clean up the already dropped partitioned data
- SegmentFileStore.cleanSegments(table, null, false)
CarbonUtils.threadSet("partition.operationcontext",
loadParams.operationContext)
val attributes = if (loadParams.scanResultRDD.isDefined) {
// take the already re-arranged attributes
@@ -1102,7 +1099,7 @@ object CommonLoadUtils {
val specs =
SegmentFileStore.getPartitionSpecs(loadParams.carbonLoadModel.getSegmentId,
loadParams.carbonLoadModel.getTablePath,
-
SegmentStatusManager.readLoadMetadata(CarbonTablePath.getMetadataPath(table.getTablePath)))
+ loadParams.carbonLoadModel.getLoadMetadataDetails.asScala.toArray)
if (specs != null) {
specs.asScala.map { spec =>
Row(spec.getPartitions.asScala.mkString("/"),
spec.getLocation.toString, spec.getUuid)
diff --git
a/integration/spark/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala
b/integration/spark/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala
index f1f0b80..225daf5 100644
---
a/integration/spark/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala
+++
b/integration/spark/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala
@@ -159,13 +159,16 @@ with Serializable {
if (currEntry != null) {
val loadEntry =
ObjectSerializationUtil.convertStringToObject(currEntry).asInstanceOf[LoadMetadataDetails]
- val details =
-
SegmentStatusManager.readLoadMetadata(CarbonTablePath.getMetadataPath(model.getTablePath))
model.setSegmentId(loadEntry.getLoadName)
model.setFactTimeStamp(loadEntry.getLoadStartTime)
- val list = new util.ArrayList[LoadMetadataDetails](details.toList.asJava)
- list.add(loadEntry)
- model.setLoadMetadataDetails(list)
+ if (!isLoadDetailsContainTheCurrentEntry(
+ model.getLoadMetadataDetails.asScala.toArray, loadEntry)) {
+ val details =
+
SegmentStatusManager.readLoadMetadata(CarbonTablePath.getMetadataPath(model.getTablePath))
+ val list = new
util.ArrayList[LoadMetadataDetails](details.toList.asJava)
+ list.add(loadEntry)
+ model.setLoadMetadataDetails(list)
+ }
}
// Set the update timestamp if user sets in case of update query. It needs
to be updated
// in load status update time
@@ -224,6 +227,14 @@ with Serializable {
}
}
override def equals(other: Any): Boolean =
other.isInstanceOf[SparkCarbonTableFormat]
+
+ private def isLoadDetailsContainTheCurrentEntry(
+ loadDetails: Array[LoadMetadataDetails],
+ currentEntry: LoadMetadataDetails): Boolean = {
+ (loadDetails.length - 1 to 0).exists { index =>
+ loadDetails(index).getLoadName.equals(currentEntry.getLoadName)
+ }
+ }
}
case class CarbonSQLHadoopMapReduceCommitProtocol(jobId: String, path: String,
isAppend: Boolean)
diff --git
a/integration/spark/src/main/scala/org/apache/spark/sql/secondaryindex/events/SILoadEventListenerForFailedSegments.scala
b/integration/spark/src/main/scala/org/apache/spark/sql/secondaryindex/events/SILoadEventListenerForFailedSegments.scala
index 2e6a441..2071385 100644
---
a/integration/spark/src/main/scala/org/apache/spark/sql/secondaryindex/events/SILoadEventListenerForFailedSegments.scala
+++
b/integration/spark/src/main/scala/org/apache/spark/sql/secondaryindex/events/SILoadEventListenerForFailedSegments.scala
@@ -64,7 +64,6 @@ class SILoadEventListenerForFailedSegments extends
OperationEventListener with L
.lookupRelation(Some(carbonLoadModel.getDatabaseName),
carbonLoadModel.getTableName)(sparkSession).asInstanceOf[CarbonRelation].carbonTable
val indexMetadata = carbonTable.getIndexMetadata
- val mainTableDetails =
SegmentStatusManager.readLoadMetadata(carbonTable.getMetadataPath)
val secondaryIndexProvider = IndexType.SI.getIndexProviderName
if (null != indexMetadata && null != indexMetadata.getIndexesMap &&
null != indexMetadata.getIndexesMap.get(secondaryIndexProvider)) {
@@ -72,6 +71,8 @@ class SILoadEventListenerForFailedSegments extends
OperationEventListener with L
.get(secondaryIndexProvider).keySet().asScala
// if there are no index tables for a given fact table do not
perform any action
if (indexTables.nonEmpty) {
+ val mainTableDetails =
+
SegmentStatusManager.readLoadMetadata(carbonTable.getMetadataPath)
indexTables.foreach {
indexTableName =>
val isLoadSIForFailedSegments =
sparkSession.sessionState.catalog
diff --git
a/processing/src/main/java/org/apache/carbondata/processing/loading/TableProcessingOperations.java
b/processing/src/main/java/org/apache/carbondata/processing/loading/TableProcessingOperations.java
index 358295d..f0f14d6 100644
---
a/processing/src/main/java/org/apache/carbondata/processing/loading/TableProcessingOperations.java
+++
b/processing/src/main/java/org/apache/carbondata/processing/loading/TableProcessingOperations.java
@@ -54,11 +54,10 @@ public class TableProcessingOperations {
public static void deletePartialLoadDataIfExist(CarbonTable carbonTable,
final boolean isCompactionFlow) throws IOException {
String metaDataLocation = carbonTable.getMetadataPath();
- final LoadMetadataDetails[] details =
SegmentStatusManager.readLoadMetadata(metaDataLocation);
-
//delete folder which metadata no exist in tablestatus
String partitionPath =
CarbonTablePath.getPartitionDir(carbonTable.getTablePath());
if (FileFactory.isFileExist(partitionPath)) {
+ final LoadMetadataDetails[] details =
SegmentStatusManager.readLoadMetadata(metaDataLocation);
CarbonFile carbonFile = FileFactory.getCarbonFile(partitionPath);
CarbonFile[] listFiles = carbonFile.listFiles(new CarbonFileFilter() {
@Override
