vikramahuja1001 commented on a change in pull request #3894:
URL:
https://github.com/apache/carbondata/pull/3894#discussion_r481107503##########
File path: integration/spark/src/main/scala/org/apache/spark/sql/secondaryindex/events/SILoadEventListenerForFailedSegments.scala
##########
@@ -57,189 +58,201 @@ class SILoadEventListenerForFailedSegments extends OperationEventListener with L
event match {
case postStatusUpdateEvent: LoadTablePostStatusUpdateEvent =>
LOGGER.info("Load post status update event-listener called")
- val loadTablePostStatusUpdateEvent = event.asInstanceOf[LoadTablePostStatusUpdateEvent]
- val carbonLoadModel = loadTablePostStatusUpdateEvent.getCarbonLoadModel
- val sparkSession = SparkSession.getActiveSession.get
- // 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
- val carbonTable = metaStore
- .lookupRelation(Some(carbonLoadModel.getDatabaseName),
- carbonLoadModel.getTableName)(sparkSession).asInstanceOf[CarbonRelation].carbonTable
- val indexMetadata = carbonTable.getIndexMetadata
- val secondaryIndexProvider = IndexType.SI.getIndexProviderName
- if (null != indexMetadata && null != indexMetadata.getIndexesMap &&
+ if (CarbonProperties.getInstance().isSIRepairEnabled) {
+ val loadTablePostStatusUpdateEvent = event.asInstanceOf[LoadTablePostStatusUpdateEvent]
+ val carbonLoadModel = loadTablePostStatusUpdateEvent.getCarbonLoadModel
+ val sparkSession = SparkSession.getActiveSession.get
+ // 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
+ val carbonTable = metaStore
+ .lookupRelation(Some(carbonLoadModel.getDatabaseName),
+ carbonLoadModel.getTableName)(sparkSession).asInstanceOf[CarbonRelation].carbonTable
+ val indexMetadata = carbonTable.getIndexMetadata
+ val secondaryIndexProvider = IndexType.SI.getIndexProviderName
+ if (null != indexMetadata && null != indexMetadata.getIndexesMap &&
null != indexMetadata.getIndexesMap.get(secondaryIndexProvider)) {
- val indexTables = indexMetadata.getIndexesMap
- .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
- .getTableMetadata(TableIdentifier(indexTableName,
- Some(carbonLoadModel.getDatabaseName))).storage.properties
- .getOrElse("isSITableEnabled", "true").toBoolean
- val indexTable = metaStore
- .lookupRelation(Some(carbonLoadModel.getDatabaseName), indexTableName)(
- sparkSession)
- .asInstanceOf[CarbonRelation]
- .carbonTable
+ val indexTables = indexMetadata.getIndexesMap
+ .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
+ .getTableMetadata(TableIdentifier(indexTableName,
+ Some(carbonLoadModel.getDatabaseName))).storage.properties
+ .getOrElse("isSITableEnabled", "true").toBoolean
+ val indexTable = metaStore
+ .lookupRelation(Some(carbonLoadModel.getDatabaseName), indexTableName)(
+ sparkSession)
+ .asInstanceOf[CarbonRelation]
+ .carbonTable
- val mainTblLoadMetadataDetails: Array[LoadMetadataDetails] =
- SegmentStatusManager.readLoadMetadata(carbonTable.getMetadataPath)
- val siTblLoadMetadataDetails: Array[LoadMetadataDetails] =
- SegmentStatusManager.readLoadMetadata(indexTable.getMetadataPath)
- var segmentLocks: ListBuffer[ICarbonLock] = ListBuffer.empty
- if (!isLoadSIForFailedSegments
+ val mainTblLoadMetadataDetails: Array[LoadMetadataDetails] =
+ SegmentStatusManager.readLoadMetadata(carbonTable.getMetadataPath)
+ val siTblLoadMetadataDetails: Array[LoadMetadataDetails] =
+ SegmentStatusManager.readLoadMetadata(indexTable.getMetadataPath)
+ var segmentLocks: ListBuffer[ICarbonLock] = ListBuffer.empty
+ if (!isLoadSIForFailedSegments
|| !CarbonInternalLoaderUtil.checkMainTableSegEqualToSISeg(
- mainTblLoadMetadataDetails,
- siTblLoadMetadataDetails)) {
- val indexColumns = indexMetadata.getIndexColumns(secondaryIndexProvider,
- indexTableName)
- val secondaryIndex = IndexModel(Some(carbonTable.getDatabaseName),
- indexMetadata.getParentTableName,
- indexColumns.split(",").toList,
- indexTableName)
+ mainTblLoadMetadataDetails,
+ siTblLoadMetadataDetails)) {
+ val indexColumns = indexMetadata.getIndexColumns(secondaryIndexProvider,
+ indexTableName)
+ val secondaryIndex = IndexModel(Some(carbonTable.getDatabaseName),
+ indexMetadata.getParentTableName,
+ indexColumns.split(",").toList,
+ indexTableName)
- var details = SegmentStatusManager.readLoadMetadata(indexTable.getMetadataPath)
- // If it empty, then no need to do further computations because the
- // tablet status might not have been created and hence next load will take care
- if (details.isEmpty) {
- return
- }
+ var details = SegmentStatusManager.readLoadMetadata(indexTable.getMetadataPath)
+ // If it empty, then no need to do further computations because the
+ // tablet status might not have been created and hence next load will take care
+ if (details.isEmpty) {
+ return
+ }
- val failedLoadMetadataDetails: java.util.List[LoadMetadataDetails] = new util
- .ArrayList[LoadMetadataDetails]()
+ val failedLoadMetadataDetails: java.util.List[LoadMetadataDetails] = new util
+ .ArrayList[LoadMetadataDetails]()
- // read the details of SI table and get all the failed segments during SI
- // creation which are MARKED_FOR_DELETE or invalid INSERT_IN_PROGRESS
- details.collect {
- case loadMetaDetail: LoadMetadataDetails =>
- if (loadMetaDetail.getSegmentStatus == SegmentStatus.MARKED_FOR_DELETE &&
+ val maxSegmentRepairLimit = CarbonProperties.getInstance().maxSIRepairLimit()
+ LOGGER.info("Number of Segments to be repaired: " + maxSegmentRepairLimit)
Review comment:
done
##########
File path: core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java
##########
@@ -2076,4 +2076,16 @@ public static boolean isAuditEnabled() {
public static void setAuditEnabled(boolean enabled) {
getInstance().addProperty(CarbonCommonConstants.CARBON_ENABLE_AUDIT, String.valueOf(enabled));
}
+
+ public boolean isSIRepairEnabled() {
+ String configuredValue = getProperty(CarbonCommonConstants.CARBON_LOAD_SI_REPAIR,
+ CarbonCommonConstants.CARBON_LOAD_SI_REPAIR_DEFAULT);
+ return Boolean.parseBoolean(configuredValue);
+ }
+
+ public int maxSIRepairLimit() {
Review comment:
done
##########
File path: core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
##########
@@ -2262,6 +2262,32 @@ private CarbonCommonConstants() {
public static final int CARBON_INDEX_SERVER_WORKER_THREADS_DEFAULT =
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
+ public static final String CARBON_SI_REPAIR_LIMIT = "carbon.si.repair.limit";
+
+ /**
+ * Default value for the number of segments to be repaired during SI repair.
+ */
+ public static final String CARBON_SI_REPAIR_LIMIT_DEFAULT = "1";
Review comment:
done
##########
File path: core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
##########
@@ -2262,6 +2262,32 @@ private CarbonCommonConstants() {
public static final int CARBON_INDEX_SERVER_WORKER_THREADS_DEFAULT =
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
+ public static final String CARBON_SI_REPAIR_LIMIT = "carbon.si.repair.limit";
Review comment:
done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[hidden email]
Locked
