added segments for prepriming in case of update success

apache · Dec 19, 2019 · 9b4ff28 · 9b4ff28
1 parent fdcfcbf
commit 9b4ff28
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 3 deletions.
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -459,20 +459,26 @@ object CarbonDataRDDFactory {
         if (resultSize == 0) {
           return null
         }
-        if (CarbonUpdateUtil.updateTableMetadataStatus(
+        if (!CarbonUpdateUtil.updateTableMetadataStatus(
           segmentDetails,
           carbonTable,
           updateModel.get.updatedTimeStamp + "",
           true,
           new util.ArrayList[Segment](0),
           new util.ArrayList[Segment](segmentFiles), "")) {
-        } else {
           LOGGER.error("Data update failed due to failure in table status updation.")
           updateModel.get.executorErrors.errorMsg = errorMessage
           updateModel.get.executorErrors.failureCauses = FailureCauses
             .STATUS_FILE_UPDATION_FAILURE
           return null
         }
+        // code to handle Pre-Priming cache for update command
+        if (!segmentFiles.isEmpty) {
+          val segmentsToPrePrime = segmentFiles.asScala.map(iterator => iterator.getSegmentNo).toSeq
+          DistributedRDDUtils
+            .triggerPrepriming(sqlContext.sparkSession, carbonTable, segmentsToPrePrime,
+              operationContext, hadoopConf, segmentsToPrePrime.toList)
+        }
       }
       return null
     }

diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
@@ -477,11 +477,12 @@ class CarbonFileMetastore extends CarbonMetaStore {
       // clear driver B-tree and dictionary cache
       ManageDictionaryAndBTree.clearBTreeAndDictionaryLRUCache(carbonTable)
     }
+    // Clear both driver and executor cache.
+    DataMapStoreManager.getInstance().clearDataMaps(absoluteTableIdentifier)
     CarbonHiveMetadataUtil.invalidateAndDropTable(dbName, tableName, sparkSession)
     // discard cached table info in cachedDataSourceTables
     val tableIdentifier = TableIdentifier(tableName, Option(dbName))
     sparkSession.sessionState.catalog.refreshTable(tableIdentifier)
-    DataMapStoreManager.getInstance().clearDataMaps(absoluteTableIdentifier)
     SegmentPropertiesAndSchemaHolder.getInstance().invalidate(absoluteTableIdentifier)
     removeTableFromMetadata(dbName, tableName)
   }