You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ku...@apache.org on 2020/09/02 08:58:39 UTC
[carbondata] branch master updated: [CARBONDATA-3957] Added
property to enable disable the repair logic and provide a limit to number
of segments in SILoadEventListenerForFailedSegments
This is an automated email from the ASF dual-hosted git repository.
kunalkapoor 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 c05b69e [CARBONDATA-3957] Added property to enable disable the repair logic and provide a limit to number of segments in SILoadEventListenerForFailedSegments
c05b69e is described below
commit c05b69e4c86f7542bc1644c9fa162d47b59df7f2
Author: Vikram Ahuja <vi...@gmail.com>
AuthorDate: Tue Aug 18 12:12:03 2020 +0530
[CARBONDATA-3957] Added property to enable disable the repair logic
and provide a limit to number of segments in SILoadEventListenerForFailedSegments
Why is this PR needed?
In the main table with SI tables after every load/insert command ,
SILoadEventListenerForFailedSegments.scala checks for missing segments or
segments mismatch in SI table and loads the missing/deleted segments to the SI
table. In case when there are very large number of missing segments in the SI
table(10000's), the repair logic will run for multiple days and will thus block the next load.
What changes were proposed in this PR?
The above mentioned issue is solved using 2 carbon properties.
1. carbon.load.si.repair- This property lets the user enable/disable the repair
SI logic in SILoadEventListenerForFailedSegments. The default value of this property is true.
2. carbon.si.repair.limit - This property decides the number of failed segments that
are being loaded again in the SI table in SILoadEventListenerForFailedSegments.
By default repairing all the segments. Instead of repairing all the segments in one
load the user can now set a limit thus not blocking the next load for the in case
of large number of missing SI segments.
This closes #3894
---
.../core/constants/CarbonCommonConstants.java | 20 ++++++
.../carbondata/core/util/CarbonProperties.java | 32 ++++++++++
.../apache/carbondata/core/util/SessionParams.java | 26 +++-----
docs/configuration-parameters.md | 2 +
.../secondaryindex/TestSIWithSecondryIndex.scala | 2 +
.../command/index/IndexRepairCommand.scala | 6 +-
.../apache/spark/sql/index/CarbonIndexUtil.scala | 71 +++++++++++-----------
.../SILoadEventListenerForFailedSegments.scala | 10 ++-
8 files changed, 116 insertions(+), 53 deletions(-)
diff --git a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
index 1a19b86..bdb3558 100644
--- a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
+++ b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
@@ -2273,6 +2273,26 @@ public final class CarbonCommonConstants {
500;
/**
+ * Configured property to enable/disable load failed segments in SI table during
+ * load/insert command.
+ */
+ @CarbonProperty(dynamicConfigurable = true)
+ public static final String CARBON_LOAD_SI_REPAIR = "carbon.load.si.repair";
+
+ /**
+ * Default value for load failed segments in SI table during
+ * load/insert command.
+ */
+ public static final String CARBON_LOAD_SI_REPAIR_DEFAULT = "true";
+
+ /**
+ * Property to give a limit to the number of segments that are reloaded in the
+ * SI table in the FailedSegments listener.
+ */
+ @CarbonProperty(dynamicConfigurable = true)
+ public static final String CARBON_SI_REPAIR_LIMIT = "carbon.si.repair.limit";
+
+ /**
* Set it to true to enable audit
*/
public static final String CARBON_ENABLE_AUDIT = "carbon.audit.enabled";
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java
index cf339d2..d0077ec 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java
@@ -2083,4 +2083,36 @@ public final class CarbonProperties {
CarbonCommonConstants.CARBON_LOAD_DATEFORMAT_SETLENIENT_ENABLE_DEFAULT);
return Boolean.parseBoolean(configuredValue);
}
+
+ public boolean isSIRepairEnabled(String dbName, String tableName) {
+ // Check if user has enabled/disabled the use of property for the current db and table using
+ // the set command
+ String configuredValue = getSessionPropertyValue(
+ CarbonCommonConstants.CARBON_LOAD_SI_REPAIR + "." + dbName + "." + tableName);
+ if (configuredValue == null) {
+ // if not set in session properties then check carbon.properties for the same.
+ configuredValue = getProperty(CarbonCommonConstants.CARBON_LOAD_SI_REPAIR,
+ CarbonCommonConstants.CARBON_LOAD_SI_REPAIR_DEFAULT);
+ }
+ boolean propertyEnabled = Boolean.parseBoolean(configuredValue);
+ if (propertyEnabled) {
+ LOGGER.info("SI Repair is enabled for " + dbName + "." + tableName);
+ }
+ return propertyEnabled;
+ }
+
+ public int getMaxSIRepairLimit(String dbName, String tableName) {
+ // Check if user has enabled/disabled the use of property for the current db and table using
+ // the set command
+ String thresholdValue = getSessionPropertyValue(
+ CarbonCommonConstants.CARBON_LOAD_SI_REPAIR + "." + dbName + "." + tableName);
+ if (thresholdValue == null) {
+ // if not set in session properties then check carbon.properties for the same.
+ thresholdValue = getProperty(CarbonCommonConstants.CARBON_SI_REPAIR_LIMIT);
+ }
+ if (thresholdValue == null) {
+ return Integer.MAX_VALUE;
+ }
+ return Math.abs(Integer.parseInt(thresholdValue));
+ }
}
diff --git a/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java b/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java
index 7a9d547..233dbe0 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java
@@ -29,22 +29,7 @@ import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.constants.CarbonLoadOptionConstants;
import org.apache.carbondata.core.exception.InvalidConfigurationException;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.CARBON_CUSTOM_BLOCK_DISTRIBUTION;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.CARBON_ENABLE_INDEX_SERVER;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.CARBON_ENABLE_MV;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.CARBON_LOAD_DATEFORMAT_SETLENIENT_ENABLE;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.CARBON_MAJOR_COMPACTION_SIZE;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.CARBON_PUSH_ROW_FILTERS_FOR_VECTOR;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.CARBON_QUERY_STAGE_INPUT;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.COMPACTION_SEGMENT_LEVEL_THRESHOLD;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.ENABLE_AUTO_LOAD_MERGE;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.ENABLE_OFFHEAP_SORT;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.ENABLE_SI_LOOKUP_PARTIALSTRING;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.ENABLE_UNSAFE_IN_QUERY_EXECUTION;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.ENABLE_UNSAFE_SORT;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.ENABLE_VECTOR_READER;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.NUM_CORES_COMPACTING;
-import static org.apache.carbondata.core.constants.CarbonCommonConstants.NUM_CORES_LOADING;
+import static org.apache.carbondata.core.constants.CarbonCommonConstants.*;
import static org.apache.carbondata.core.constants.CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_ACTION;
import static org.apache.carbondata.core.constants.CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_LOGGER_ENABLE;
import static org.apache.carbondata.core.constants.CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORD_PATH;
@@ -155,6 +140,7 @@ public class SessionParams implements Serializable, Cloneable {
case ENABLE_AUTO_LOAD_MERGE:
case CARBON_PUSH_ROW_FILTERS_FOR_VECTOR:
case CARBON_LOAD_DATEFORMAT_SETLENIENT_ENABLE:
+ case CARBON_LOAD_SI_REPAIR:
case CARBON_ENABLE_INDEX_SERVER:
case CARBON_QUERY_STAGE_INPUT:
case CARBON_ENABLE_MV:
@@ -181,6 +167,7 @@ public class SessionParams implements Serializable, Cloneable {
break;
case CARBON_OPTIONS_GLOBAL_SORT_PARTITIONS:
case NUM_CORES_LOADING:
+ case CARBON_SI_REPAIR_LIMIT:
case NUM_CORES_COMPACTING:
case BLOCKLET_SIZE_IN_MB:
case CARBON_MAJOR_COMPACTION_SIZE:
@@ -215,6 +202,13 @@ public class SessionParams implements Serializable, Cloneable {
default:
if (key.startsWith(CARBON_ENABLE_INDEX_SERVER) && key.split("\\.").length == 6) {
isValid = true;
+ } else if (key.startsWith(CARBON_SI_REPAIR_LIMIT)) {
+ isValid = CarbonUtil.validateValidIntType(value);
+ if (!isValid) {
+ throw new InvalidConfigurationException("Invalid CARBON_SI_REPAIR_LIMIT");
+ }
+ } else if (key.startsWith(CARBON_LOAD_SI_REPAIR) && key.split("\\.").length == 6) {
+ isValid = true;
} else if (key.startsWith(CarbonCommonConstants.CARBON_INPUT_SEGMENTS)) {
isValid = CarbonUtil.validateRangeOfSegmentList(value);
if (!isValid) {
diff --git a/docs/configuration-parameters.md b/docs/configuration-parameters.md
index 3e4f8bd..634d607 100644
--- a/docs/configuration-parameters.md
+++ b/docs/configuration-parameters.md
@@ -94,6 +94,8 @@ This section provides the details of all the configurations required for the Car
| carbon.binary.decoder | None | Support configurable decode for loading. Two decoders supported: base64 and hex |
| carbon.local.dictionary.size.threshold.inmb | 4 | size based threshold for local dictionary in MB, maximum allowed size is 16 MB. |
| carbon.enable.bad.record.handling.for.insert | false | by default, disable the bad record and converter step during "insert into" |
+| carbon.load.si.repair | true | by default, enable loading for failed segments in SI during load/insert command |
+| carbon.si.repair.limit | (none) | Number of failed segments to be loaded in SI when repairing missing segments in SI, by default load all the missing segments. Supports value from 0 to 2147483646 |
## Compaction Configuration
diff --git a/index/secondary-index/src/test/scala/org/apache/carbondata/spark/testsuite/secondaryindex/TestSIWithSecondryIndex.scala b/index/secondary-index/src/test/scala/org/apache/carbondata/spark/testsuite/secondaryindex/TestSIWithSecondryIndex.scala
index 53333f0..713047e 100644
--- a/index/secondary-index/src/test/scala/org/apache/carbondata/spark/testsuite/secondaryindex/TestSIWithSecondryIndex.scala
+++ b/index/secondary-index/src/test/scala/org/apache/carbondata/spark/testsuite/secondaryindex/TestSIWithSecondryIndex.scala
@@ -175,6 +175,7 @@ class TestSIWithSecondryIndex extends QueryTest with BeforeAndAfterAll {
val carbontable = CarbonEnv.getCarbonTable(Some("default"), "uniqdata")(sqlContext.sparkSession)
val details = SegmentStatusManager.readLoadMetadata(indexTable.getMetadataPath)
val failSegments = List("3","4")
+ sql(s"""set carbon.si.repair.limit = 2""")
var loadMetadataDetailsList = Array[LoadMetadataDetails]()
details.foreach{detail =>
if(failSegments.contains(detail.getLoadName)){
@@ -201,6 +202,7 @@ class TestSIWithSecondryIndex extends QueryTest with BeforeAndAfterAll {
|SERDEPROPERTIES ('isSITableEnabled' = 'false')""".stripMargin)
val count2 = sql("select * from uniqdata where workgroupcategoryname = 'developer'").count()
val df2 = sql("select * from uniqdata where workgroupcategoryname = 'developer'").queryExecution.sparkPlan
+ sql(s"""set carbon.si.repair.limit = 1""")
assert(count1 == count2)
assert(isFilterPushedDownToSI(df1))
assert(!isFilterPushedDownToSI(df2))
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/index/IndexRepairCommand.scala b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/index/IndexRepairCommand.scala
index 6a3e4ec..43efba8 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/index/IndexRepairCommand.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/index/IndexRepairCommand.scala
@@ -112,7 +112,8 @@ extends DataCommand {
indexTables.foreach {
indexTableName =>
CarbonIndexUtil.processSIRepair(indexTableName, mainCarbonTable, carbonLoadModel,
- indexMetadata, mainTableDetails, secondaryIndexProvider)(sparkSession)
+ indexMetadata, mainTableDetails, secondaryIndexProvider,
+ Integer.MAX_VALUE)(sparkSession)
}
} else {
val indexTablesToRepair = indexTables.filter(indexTable => indexTable
@@ -120,7 +121,8 @@ extends DataCommand {
indexTablesToRepair.foreach {
indexTableName =>
CarbonIndexUtil.processSIRepair(indexTableName, mainCarbonTable, carbonLoadModel,
- indexMetadata, mainTableDetails, secondaryIndexProvider)(sparkSession)
+ indexMetadata, mainTableDetails, secondaryIndexProvider,
+ Integer.MAX_VALUE)(sparkSession)
}
if (indexTablesToRepair.isEmpty) {
throw new Exception("Unable to find index table" + indexTableToRepair.get)
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/index/CarbonIndexUtil.scala b/integration/spark/src/main/scala/org/apache/spark/sql/index/CarbonIndexUtil.scala
index c2bb9cf..0592808 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/index/CarbonIndexUtil.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/index/CarbonIndexUtil.scala
@@ -383,7 +383,8 @@ object CarbonIndexUtil {
def processSIRepair(indexTableName: String, carbonTable: CarbonTable,
carbonLoadModel: CarbonLoadModel, indexMetadata: IndexMetadata,
- mainTableDetails: List[LoadMetadataDetails], secondaryIndexProvider: String)
+ mainTableDetails: List[LoadMetadataDetails], secondaryIndexProvider: String,
+ repairLimit: Int)
(sparkSession: SparkSession) : Unit = {
// when Si creation and load to main table are parallel, get the carbonTable from the
// metastore which will have the latest index Info
@@ -423,14 +424,14 @@ object CarbonIndexUtil {
case loadMetaDetail: LoadMetadataDetails =>
if (loadMetaDetail.getSegmentStatus == SegmentStatus.MARKED_FOR_DELETE &&
checkIfMainTableLoadIsValid(mainTableDetails.toArray,
- loadMetaDetail.getLoadName)) {
+ loadMetaDetail.getLoadName) && repairLimit > failedLoadMetadataDetails.size() ) {
failedLoadMetadataDetails.add(loadMetaDetail)
} else if ((loadMetaDetail.getSegmentStatus ==
SegmentStatus.INSERT_IN_PROGRESS ||
loadMetaDetail.getSegmentStatus ==
SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS) &&
checkIfMainTableLoadIsValid(mainTableDetails.toArray,
- loadMetaDetail.getLoadName)) {
+ loadMetaDetail.getLoadName) && repairLimit > failedLoadMetadataDetails.size()) {
val segmentLock = CarbonLockFactory
.getCarbonLockObj(indexTable.getAbsoluteTableIdentifier,
CarbonTablePath.addSegmentPrefix(loadMetaDetail.getLoadName) +
@@ -454,37 +455,39 @@ object CarbonIndexUtil {
// status file and get the skipped segments if any
CarbonInternalLoaderUtil.getListOfValidSlices(mainTableDetails.toArray).asScala
.foreach(metadataDetail => {
- val detail = siTblLoadMetadataDetails
- .filter(metadata => metadata.getLoadName.equals(metadataDetail))
- val mainTableDetail = mainTableDetails
- .filter(metadata => metadata.getLoadName.equals(metadataDetail))
- if (null == detail || detail.length == 0) {
- val newDetails = new LoadMetadataDetails
- newDetails.setLoadName(metadataDetail)
- LOGGER.error("Added in SILoadFailedSegment " + newDetails.getLoadName + " for SI" +
- " table " + indexTableName + "." + carbonTable.getTableName )
- failedLoadMetadataDetails.add(newDetails)
- } else if (detail != null && detail.length != 0 && metadataDetail != null
- && metadataDetail.length != 0) {
- // If SI table has compacted segments and main table does not have
- // compacted segments due to some failure while compaction, need to
- // reload the original segments in this case.
- if (detail(0).getSegmentStatus == SegmentStatus.COMPACTED &&
- mainTableDetail(0).getSegmentStatus == SegmentStatus.SUCCESS) {
- detail(0).setSegmentStatus(SegmentStatus.SUCCESS)
- // in concurrent scenario, if a compaction is going on table, then SI
- // segments are updated first in table status and then the main table
- // segment, so in any load runs parallel this listener shouldn't consider
- // those segments accidentally. So try to take the segment lock.
- val segmentLockOfProbableOnCompactionSeg = CarbonLockFactory
- .getCarbonLockObj(carbonTable.getAbsoluteTableIdentifier,
- CarbonTablePath.addSegmentPrefix(mainTableDetail(0).getLoadName) +
- LockUsage.LOCK)
- if (segmentLockOfProbableOnCompactionSeg.lockWithRetries()) {
- segmentLocks += segmentLockOfProbableOnCompactionSeg
- LOGGER.error("Added in SILoadFailedSegment " + detail(0).getLoadName + " for SI " +
- "table " + indexTableName + "." + carbonTable.getTableName )
- failedLoadMetadataDetails.add(detail(0))
+ if (repairLimit > failedLoadMetadataDetails.size()) {
+ val detail = siTblLoadMetadataDetails
+ .filter(metadata => metadata.getLoadName.equals(metadataDetail))
+ val mainTableDetail = mainTableDetails
+ .filter(metadata => metadata.getLoadName.equals(metadataDetail))
+ if (null == detail || detail.length == 0) {
+ val newDetails = new LoadMetadataDetails
+ newDetails.setLoadName(metadataDetail)
+ LOGGER.error("Added in SILoadFailedSegment " + newDetails.getLoadName + " for SI" +
+ " table " + indexTableName + "." + carbonTable.getTableName)
+ failedLoadMetadataDetails.add(newDetails)
+ } else if (detail != null && detail.length != 0 && metadataDetail != null
+ && metadataDetail.length != 0) {
+ // If SI table has compacted segments and main table does not have
+ // compacted segments due to some failure while compaction, need to
+ // reload the original segments in this case.
+ if (detail(0).getSegmentStatus == SegmentStatus.COMPACTED &&
+ mainTableDetail(0).getSegmentStatus == SegmentStatus.SUCCESS) {
+ detail(0).setSegmentStatus(SegmentStatus.SUCCESS)
+ // in concurrent scenario, if a compaction is going on table, then SI
+ // segments are updated first in table status and then the main table
+ // segment, so in any load runs parallel this listener shouldn't consider
+ // those segments accidentally. So try to take the segment lock.
+ val segmentLockOfProbableOnCompactionSeg = CarbonLockFactory
+ .getCarbonLockObj(carbonTable.getAbsoluteTableIdentifier,
+ CarbonTablePath.addSegmentPrefix(mainTableDetail(0).getLoadName) +
+ LockUsage.LOCK)
+ if (segmentLockOfProbableOnCompactionSeg.lockWithRetries()) {
+ segmentLocks += segmentLockOfProbableOnCompactionSeg
+ LOGGER.error("Added in SILoadFailedSegment " + detail(0).getLoadName + " for SI "
+ + "table " + indexTableName + "." + carbonTable.getTableName)
+ failedLoadMetadataDetails.add(detail(0))
+ }
}
}
}
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 cd7294b..a1e6fc1 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
@@ -49,6 +49,8 @@ class SILoadEventListenerForFailedSegments extends OperationEventListener with L
val loadTablePostStatusUpdateEvent = event.asInstanceOf[LoadTablePostStatusUpdateEvent]
val carbonLoadModel = loadTablePostStatusUpdateEvent.getCarbonLoadModel
val sparkSession = SparkSession.getActiveSession.get
+ if (CarbonProperties.getInstance().isSIRepairEnabled(carbonLoadModel.getDatabaseName,
+ carbonLoadModel.getTableName)) {
// when Si creation and load to main table are parallel, get the carbonTable from the
// metastore which will have the latest index Info
val metaStore = CarbonEnv.getInstance(sparkSession).carbonMetaStore
@@ -62,6 +64,10 @@ class SILoadEventListenerForFailedSegments extends OperationEventListener with L
null != indexMetadata.getIndexesMap.get(secondaryIndexProvider)) {
val indexTables = indexMetadata.getIndexesMap
.get(secondaryIndexProvider).keySet().asScala
+ val maxSegmentRepairLimit = CarbonProperties.getInstance().getMaxSIRepairLimit(
+ carbonLoadModel.getDatabaseName, carbonLoadModel.getTableName)
+ LOGGER.info("Number of segments to be repaired for table: " +
+ carbonLoadModel.getTableName + " are : " + maxSegmentRepairLimit)
// if there are no index tables for a given fact table do not perform any action
if (indexTables.nonEmpty) {
val mainTableDetails =
@@ -69,10 +75,12 @@ class SILoadEventListenerForFailedSegments extends OperationEventListener with L
indexTables.foreach {
indexTableName =>
CarbonIndexUtil.processSIRepair(indexTableName, carbonTable, carbonLoadModel,
- indexMetadata, mainTableDetails.toList, secondaryIndexProvider)(sparkSession)
+ indexMetadata, mainTableDetails.toList, secondaryIndexProvider,
+ maxSegmentRepairLimit)(sparkSession)
}
}
}
+ }
}
}
}