ajantha-bhat commented on a change in pull request #4110:
URL: https://github.com/apache/carbondata/pull/4110#discussion_r607816175
##########
File path:
integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
##########
@@ -128,6 +129,16 @@ class CarbonScanRDD[T: ClassTag](
} else {
prepareInputFormatForDriver(job.getConfiguration)
}
+ if (segmentIds.isDefined) {
+ CarbonInputFormat.setQuerySegment(job.getConfiguration, segmentIds.get)
Review comment:
we already have `segmentsToAccess`, we need one more variable?
##########
File path:
integration/spark/src/main/scala/org/apache/spark/sql/execution/command/index/CarbonCreateIndexCommand.scala
##########
@@ -225,7 +225,9 @@ case class CarbonCreateIndexCommand(
new IndexTableInfo(parentTable.getDatabaseName,
indexModel.indexName,
indexSchema.getProperties),
false)
- val enabledIndexInfo = IndexTableInfo.enableIndex(indexInfo,
indexModel.indexName)
+ val enabledIndexInfo = IndexTableInfo.setIndexStatus(indexInfo,
Review comment:
For the existing SI table, what will be the behavior? can be queried
from presto?
better to update the document on this.
##########
File path:
integration/spark/src/main/scala/org/apache/spark/sql/index/CarbonIndexUtil.scala
##########
@@ -396,6 +398,88 @@ object CarbonIndexUtil {
}
}
+ def updateIndexStatusInBatch(carbonTable: CarbonTable,
+ indexTables: List[CarbonTable],
+ indexType: IndexType,
+ status: IndexStatus,
+ sparkSession: SparkSession): Unit = {
+ val dbName = carbonTable.getDatabaseName
+ val tableName = carbonTable.getTableName
+ val metadataLock =
CarbonLockFactory.getCarbonLockObj(carbonTable.getAbsoluteTableIdentifier,
+ LockUsage.METADATA_LOCK)
+ try {
+ if (metadataLock.lockWithRetries()) {
+ CarbonMetadata.getInstance.removeTable(dbName, tableName)
+ val table = CarbonEnv.getCarbonTable(Some(dbName),
tableName)(sparkSession)
+ val indexTableInfos = IndexTableInfo.fromGson(table.getIndexInfo)
+ val indexMetadata = table.getIndexMetadata
+ indexTables.foreach { indexTable =>
+ IndexTableInfo.setIndexStatus(indexTableInfos,
indexTable.getTableName, status)
+ indexMetadata.updateIndexStatus(indexType.getIndexProviderName,
+ indexTable.getTableName,
+ status.name())
+ }
+ table.getTableInfo
+ .getFactTable
+ .getTableProperties
+ .put(table.getCarbonTableIdentifier.getTableId,
indexMetadata.serialize)
+ sparkSession.sql(s"""ALTER TABLE $dbName.$tableName SET
SERDEPROPERTIES ('indexInfo' = '${
+ IndexTableInfo.toGson(indexTableInfos)
+ }')""".stripMargin).collect()
+ CarbonHiveIndexMetadataUtil.refreshTable(dbName, tableName,
sparkSession)
+ CarbonMetadata.getInstance.removeTable(dbName, tableName)
+ CarbonMetadata.getInstance.loadTableMetadata(table.getTableInfo)
+ }
+ } finally {
+ metadataLock.unlock()
+ }
+ }
+
+ def updateIndexStatus(carbonTable: CarbonTable,
+ indexName: String,
+ indexType: IndexType,
+ status: IndexStatus,
+ needLock: Boolean = true,
+ sparkSession: SparkSession): Unit = {
+ val dbName = carbonTable.getDatabaseName
Review comment:
we cannot pass the single table to the above batch API and update only
that, why do we need similar functionality again ?
If both needed please extract common code and reuse
##########
File path:
processing/src/main/java/org/apache/carbondata/processing/index/IndexWriterListener.java
##########
@@ -75,7 +76,8 @@ public void registerAllWriter(CarbonTable carbonTable, String
segmentId,
for (TableIndex tableIndex : tableIndices) {
// register it only if it is not lazy index, for lazy index, user
// will rebuild the index manually
Review comment:
please update the comments
##########
File path:
integration/spark/src/main/scala/org/apache/spark/sql/index/CarbonIndexUtil.scala
##########
@@ -396,6 +398,88 @@ object CarbonIndexUtil {
}
}
+ def updateIndexStatusInBatch(carbonTable: CarbonTable,
+ indexTables: List[CarbonTable],
+ indexType: IndexType,
+ status: IndexStatus,
+ sparkSession: SparkSession): Unit = {
+ val dbName = carbonTable.getDatabaseName
+ val tableName = carbonTable.getTableName
+ val metadataLock =
CarbonLockFactory.getCarbonLockObj(carbonTable.getAbsoluteTableIdentifier,
+ LockUsage.METADATA_LOCK)
+ try {
+ if (metadataLock.lockWithRetries()) {
+ CarbonMetadata.getInstance.removeTable(dbName, tableName)
+ val table = CarbonEnv.getCarbonTable(Some(dbName),
tableName)(sparkSession)
+ val indexTableInfos = IndexTableInfo.fromGson(table.getIndexInfo)
+ val indexMetadata = table.getIndexMetadata
+ indexTables.foreach { indexTable =>
+ IndexTableInfo.setIndexStatus(indexTableInfos,
indexTable.getTableName, status)
+ indexMetadata.updateIndexStatus(indexType.getIndexProviderName,
+ indexTable.getTableName,
+ status.name())
+ }
+ table.getTableInfo
+ .getFactTable
+ .getTableProperties
+ .put(table.getCarbonTableIdentifier.getTableId,
indexMetadata.serialize)
+ sparkSession.sql(s"""ALTER TABLE $dbName.$tableName SET
SERDEPROPERTIES ('indexInfo' = '${
+ IndexTableInfo.toGson(indexTableInfos)
+ }')""".stripMargin).collect()
+ CarbonHiveIndexMetadataUtil.refreshTable(dbName, tableName,
sparkSession)
+ CarbonMetadata.getInstance.removeTable(dbName, tableName)
+ CarbonMetadata.getInstance.loadTableMetadata(table.getTableInfo)
+ }
+ } finally {
+ metadataLock.unlock()
+ }
+ }
+
+ def updateIndexStatus(carbonTable: CarbonTable,
Review comment:
agree. Maybe we should support this in the refresh SI command with the
option to set this for existing tables.
--
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:
[email protected]
