You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ra...@apache.org on 2018/05/06 08:50:16 UTC
[1/4] carbondata git commit: [CARBONDATA-2415] Support Refresh
DataMap command for all Index datamap
Repository: carbondata
Updated Branches:
refs/heads/master f2fb06806 -> 9db662a2d
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark2/src/main/scala/org/apache/carbondata/datamap/lucene/LuceneDataMapRefreshRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/datamap/lucene/LuceneDataMapRefreshRDD.scala b/integration/spark2/src/main/scala/org/apache/carbondata/datamap/lucene/LuceneDataMapRefreshRDD.scala
deleted file mode 100644
index 2bc3eaf..0000000
--- a/integration/spark2/src/main/scala/org/apache/carbondata/datamap/lucene/LuceneDataMapRefreshRDD.scala
+++ /dev/null
@@ -1,299 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.carbondata.datamap.lucene
-
-import java.text.SimpleDateFormat
-import java.util
-
-import scala.collection.JavaConverters._
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.mapred.JobConf
-import org.apache.hadoop.mapreduce.{Job, TaskAttemptID, TaskType}
-import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
-import org.apache.spark.{CarbonInputMetrics, Partition, SparkContext, TaskContext}
-import org.apache.spark.deploy.SparkHadoopUtil
-import org.apache.spark.sql.SparkSession
-
-import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.datamap.Segment
-import org.apache.carbondata.core.datastore.impl.FileFactory
-import org.apache.carbondata.core.metadata.datatype.{DataType, DataTypes}
-import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema, TableInfo}
-import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn
-import org.apache.carbondata.core.statusmanager.SegmentStatusManager
-import org.apache.carbondata.core.util.TaskMetricsMap
-import org.apache.carbondata.core.util.path.CarbonTablePath
-import org.apache.carbondata.hadoop.{CarbonInputSplit, CarbonMultiBlockSplit, CarbonProjection, CarbonRecordReader}
-import org.apache.carbondata.hadoop.api.{CarbonInputFormat, CarbonTableInputFormat}
-import org.apache.carbondata.hadoop.readsupport.CarbonReadSupport
-import org.apache.carbondata.spark.{RefreshResult, RefreshResultImpl}
-import org.apache.carbondata.spark.rdd.{CarbonRDDWithTableInfo, CarbonSparkPartition}
-import org.apache.carbondata.spark.util.SparkDataTypeConverterImpl
-
-object LuceneDataMapRefreshRDD {
-
- def refreshDataMap(
- sparkSession: SparkSession,
- carbonTable: CarbonTable,
- schema: DataMapSchema
- ): Unit = {
- val tableIdentifier = carbonTable.getAbsoluteTableIdentifier
- val segmentStatusManager = new SegmentStatusManager(tableIdentifier)
- val validAndInvalidSegments = segmentStatusManager.getValidAndInvalidSegments()
- val validSegments = validAndInvalidSegments.getValidSegments()
- val indexedCarbonColumns =
- LuceneDataMapFactoryBase.validateAndGetIndexedColumns(schema, carbonTable)
-
- // loop all segments to rebuild DataMap
- val tableInfo = carbonTable.getTableInfo()
- validSegments.asScala.foreach { segment =>
- val dataMapStorePath = LuceneDataMapWriter.genDataMapStorePath(
- tableIdentifier.getTablePath(),
- segment.getSegmentNo(),
- schema.getDataMapName)
- // if lucene datamap folder is exists, not require to build lucene datamap again
- refreshOneSegment(sparkSession, tableInfo, dataMapStorePath, schema.getDataMapName,
- indexedCarbonColumns, segment.getSegmentNo());
- }
- }
-
- def refreshOneSegment(
- sparkSession: SparkSession,
- tableInfo: TableInfo,
- dataMapStorePath: String,
- dataMapName: String,
- indexColumns: java.util.List[String],
- segmentId: String): Unit = {
-
- if (!FileFactory.isFileExist(dataMapStorePath)) {
- if (FileFactory.mkdirs(dataMapStorePath, FileFactory.getFileType(dataMapStorePath))) {
- try {
- val status = new LuceneDataMapRefreshRDD[String, Boolean](
- sparkSession.sparkContext,
- new RefreshResultImpl(),
- tableInfo,
- dataMapStorePath,
- dataMapName,
- indexColumns.asScala.toArray,
- segmentId
- ).collect()
-
- status.find(_._2 == false).foreach { task =>
- throw new Exception(
- s"Task Failed to refresh datamap $dataMapName on segment_$segmentId")
- }
- } catch {
- case ex =>
- // process failure
- FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(dataMapStorePath))
- throw new Exception(
- s"Failed to refresh datamap $dataMapName on segment_$segmentId", ex)
- }
- }
- }
- }
-
-}
-
-class OriginalReadSupport(dataTypes: Array[DataType]) extends CarbonReadSupport[Array[Object]] {
- override def initialize(carbonColumns: Array[CarbonColumn],
- carbonTable: CarbonTable): Unit = {
- }
-
- override def readRow(data: Array[Object]): Array[Object] = {
- for (i <- 0 until dataTypes.length) {
- if (dataTypes(i) == DataTypes.STRING) {
- data(i) = data(i).toString
- }
- }
- data
- }
-
- override def close(): Unit = {
- }
-}
-
-class LuceneDataMapRefreshRDD[K, V](
- sc: SparkContext,
- result: RefreshResult[K, V],
- @transient tableInfo: TableInfo,
- dataMapStorePath: String,
- dataMapName: String,
- indexColumns: Array[String],
- segmentId: String
-) extends CarbonRDDWithTableInfo[(K, V)](sc, Nil, tableInfo.serialize()) {
-
- private val queryId = sparkContext.getConf.get("queryId", System.nanoTime() + "")
-
- private val jobTrackerId: String = {
- val formatter = new SimpleDateFormat("yyyyMMddHHmm")
- formatter.format(new util.Date())
- }
-
- override def internalCompute(split: Partition, context: TaskContext): Iterator[(K, V)] = {
- val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
- var status = false
- val inputMetrics = new CarbonInputMetrics
- TaskMetricsMap.getInstance().registerThreadCallback()
- val inputSplit = split.asInstanceOf[CarbonSparkPartition].split.value
- inputMetrics.initBytesReadCallback(context, inputSplit)
-
- val attemptId = new TaskAttemptID(jobTrackerId, id, TaskType.MAP, split.index, 0)
- val attemptContext = new TaskAttemptContextImpl(new Configuration(), attemptId)
- val format = createInputFormat(attemptContext)
-
- val taskName = CarbonTablePath.getUniqueTaskName(inputSplit.getAllSplits.get(0).getBlockPath)
-
- val tableInfo = getTableInfo
- val identifier = tableInfo.getOrCreateAbsoluteTableIdentifier()
-
- val columns = tableInfo.getFactTable.getListOfColumns.asScala
- val dataTypes = indexColumns.map { columnName =>
- columns.find(_.getColumnName.equals(columnName)).get.getDataType
- }
-
- val indexPath = LuceneDataMapWriter.genDataMapStorePathOnTaskId(identifier.getTablePath,
- segmentId, dataMapName, taskName)
-
- val model = format.createQueryModel(inputSplit, attemptContext)
- // one query id per table
- model.setQueryId(queryId)
- model.setVectorReader(false)
- model.setForcedDetailRawQuery(false)
- model.setRequiredRowId(true)
- var reader: CarbonRecordReader[Array[Object]] = null
- var indexBuilder: LuceneIndexRefreshBuilder = null
- try {
- reader = new CarbonRecordReader(model, new OriginalReadSupport(dataTypes), inputMetrics)
- reader.initialize(inputSplit, attemptContext)
-
- indexBuilder = new LuceneIndexRefreshBuilder(indexPath, indexColumns, dataTypes)
- indexBuilder.initialize()
-
- while (reader.nextKeyValue()) {
- indexBuilder.addDocument(reader.getCurrentValue)
- }
-
- indexBuilder.finish()
-
- status = true
- } finally {
- if (reader != null) {
- try {
- reader.close()
- } catch {
- case ex =>
- LOGGER.error(ex, "Failed to close reader")
- }
- }
-
- if (indexBuilder != null) {
- try {
- indexBuilder.close()
- } catch {
- case ex =>
- LOGGER.error(ex, "Failed to close index writer")
- }
- }
- }
-
- new Iterator[(K, V)] {
-
- var finished = false
-
- override def hasNext: Boolean = {
- !finished
- }
-
- override def next(): (K, V) = {
- finished = true
- result.getKey(split.index.toString, status)
- }
- }
- }
-
-
- private def createInputFormat(
- attemptContext: TaskAttemptContextImpl) = {
- val format = new CarbonTableInputFormat[Object]
- val tableInfo1 = getTableInfo
- val conf = attemptContext.getConfiguration
- CarbonInputFormat.setTableInfo(conf, tableInfo1)
- CarbonInputFormat.setDatabaseName(conf, tableInfo1.getDatabaseName)
- CarbonInputFormat.setTableName(conf, tableInfo1.getFactTable.getTableName)
- CarbonInputFormat.setDataTypeConverter(conf, classOf[SparkDataTypeConverterImpl])
-
- val identifier = tableInfo1.getOrCreateAbsoluteTableIdentifier()
- CarbonInputFormat.setTablePath(
- conf,
- identifier.appendWithLocalPrefix(identifier.getTablePath))
-
- CarbonInputFormat.setSegmentsToAccess(
- conf,
- Segment.toSegmentList(Array(segmentId), null))
-
- CarbonInputFormat.setColumnProjection(
- conf,
- new CarbonProjection(indexColumns))
- format
- }
-
- override protected def getPartitions = {
- val conf = new Configuration()
- val jobConf = new JobConf(conf)
- SparkHadoopUtil.get.addCredentials(jobConf)
- val job = Job.getInstance(jobConf)
- job.getConfiguration.set("query.id", queryId)
-
- val format = new CarbonTableInputFormat[Object]
-
- CarbonInputFormat.setSegmentsToAccess(
- job.getConfiguration,
- Segment.toSegmentList(Array(segmentId), null))
-
- CarbonInputFormat.setTableInfo(
- job.getConfiguration,
- tableInfo)
- CarbonInputFormat.setTablePath(
- job.getConfiguration,
- tableInfo.getOrCreateAbsoluteTableIdentifier().getTablePath)
- CarbonInputFormat.setDatabaseName(
- job.getConfiguration,
- tableInfo.getDatabaseName)
- CarbonInputFormat.setTableName(
- job.getConfiguration,
- tableInfo.getFactTable.getTableName)
-
- format
- .getSplits(job)
- .asScala
- .map(_.asInstanceOf[CarbonInputSplit])
- .groupBy(_.taskId)
- .map { group =>
- new CarbonMultiBlockSplit(
- group._2.asJava,
- group._2.flatMap(_.getLocations).toArray)
- }
- .zipWithIndex
- .map { split =>
- new CarbonSparkPartition(id, split._2, split._1)
- }
- .toArray
- }
-}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
index 3fe2c00..34a4013 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.execution.command.datamap
import scala.collection.JavaConverters._
+import scala.collection.mutable
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
@@ -24,11 +25,11 @@ import org.apache.spark.sql.execution.command._
import org.apache.carbondata.common.exceptions.sql.{MalformedCarbonCommandException, MalformedDataMapCommandException}
import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.datamap.{DataMapProvider, DataMapStoreManager, IndexDataMapProvider}
+import org.apache.carbondata.core.datamap.{DataMapProvider, DataMapStoreManager}
import org.apache.carbondata.core.datamap.status.DataMapStatusManager
import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider
import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}
-import org.apache.carbondata.datamap.DataMapManager
+import org.apache.carbondata.datamap.{DataMapManager, IndexDataMapProvider}
/**
* Below command class will be used to create datamap on table
@@ -37,7 +38,7 @@ import org.apache.carbondata.datamap.DataMapManager
case class CarbonCreateDataMapCommand(
dataMapName: String,
tableIdentifier: Option[TableIdentifier],
- dmClassName: String,
+ dmProviderName: String,
dmProperties: Map[String, String],
queryString: Option[String],
ifNotExistsSet: Boolean = false)
@@ -68,41 +69,39 @@ case class CarbonCreateDataMapCommand(
}
}
- dataMapSchema = new DataMapSchema(dataMapName, dmClassName)
- // TODO: move this if logic inside lucene module
- if (dataMapSchema.getProviderName.equalsIgnoreCase(DataMapClassProvider.LUCENE.getShortName) ||
- dataMapSchema.getProviderName.equalsIgnoreCase(DataMapClassProvider.LUCENE.getClassName)) {
- val datamaps = DataMapStoreManager.getInstance().getAllDataMap(mainTable).asScala
- if (datamaps.nonEmpty) {
- datamaps.foreach(datamap => {
- val dmColumns = datamap.getDataMapSchema.getProperties.get("text_columns").trim
- .toLowerCase.split(",").toSet
- val existingColumns = dmProperties("text_columns").trim.toLowerCase().split(",").toSet
- val duplicateDMColumn = dmColumns.intersect(existingColumns)
- if (duplicateDMColumn.nonEmpty) {
- throw new MalformedDataMapCommandException(
- s"Create lucene datamap $dataMapName failed, datamap already exists on column(s) " +
- s"$duplicateDMColumn")
- }
- })
- }
- }
if (mainTable != null &&
mainTable.isStreamingTable &&
- !(dataMapSchema.getProviderName.equalsIgnoreCase(DataMapClassProvider.PREAGGREGATE.toString)
- || dataMapSchema.getProviderName
- .equalsIgnoreCase(DataMapClassProvider.TIMESERIES.toString))) {
- throw new MalformedCarbonCommandException(s"Streaming table does not support creating ${
- dataMapSchema.getProviderName
- } datamap")
+ !(dmProviderName.equalsIgnoreCase(DataMapClassProvider.PREAGGREGATE.toString)
+ || dmProviderName.equalsIgnoreCase(DataMapClassProvider.TIMESERIES.toString))) {
+ throw new MalformedCarbonCommandException(s"Streaming table does not support creating " +
+ s"$dmProviderName datamap")
}
+
+ dataMapSchema = new DataMapSchema(dataMapName, dmProviderName)
dataMapSchema.setProperties(new java.util.HashMap[String, String](
dmProperties.map(x => (x._1.trim, x._2.trim)).asJava))
- dataMapProvider = DataMapManager.get().getDataMapProvider(dataMapSchema, sparkSession)
- dataMapProvider.initMeta(mainTable, dataMapSchema, queryString.orNull)
- // TODO Currently this feature is only available for index datamaps
- if (dataMapProvider.isInstanceOf[IndexDataMapProvider]) {
- DataMapStatusManager.disableDataMap(dataMapName)
+
+ dataMapProvider = DataMapManager.get.getDataMapProvider(mainTable, dataMapSchema, sparkSession)
+
+ // If it is index datamap, check whether the column has datamap created already
+ dataMapProvider match {
+ case provider: IndexDataMapProvider =>
+ val datamaps = DataMapStoreManager.getInstance.getAllDataMap(mainTable).asScala
+ val existingIndexColumn = mutable.Set[String]()
+ datamaps.foreach { datamap =>
+ datamap.getDataMapSchema.getIndexColumns.foreach(existingIndexColumn.add)
+ }
+
+ provider.getIndexedColumns.asScala.foreach { column =>
+ if (existingIndexColumn.contains(column.getColName)) {
+ throw new MalformedDataMapCommandException(String.format(
+ "column '%s' already has datamap created", column.getColName))
+ }
+ }
+ dataMapProvider.initMeta(queryString.orNull)
+ DataMapStatusManager.disableDataMap(dataMapName)
+ case _ =>
+ dataMapProvider.initMeta(queryString.orNull)
}
val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
LOGGER.audit(s"DataMap $dataMapName successfully added")
@@ -111,11 +110,11 @@ case class CarbonCreateDataMapCommand(
override def processData(sparkSession: SparkSession): Seq[Row] = {
if (dataMapProvider != null) {
- dataMapProvider.initData(mainTable)
- if (mainTable != null && mainTable.isAutoRefreshDataMap) {
- if (!DataMapClassProvider.LUCENE.getShortName.equals(dataMapSchema.getProviderName)) {
- dataMapProvider.rebuild(mainTable, dataMapSchema)
- }
+ dataMapProvider.initData()
+ if (mainTable != null &&
+ mainTable.isAutoRefreshDataMap &&
+ !dataMapSchema.isIndexDataMap) {
+ dataMapProvider.rebuild()
}
}
Seq.empty
@@ -123,7 +122,7 @@ case class CarbonCreateDataMapCommand(
override def undoMetadata(sparkSession: SparkSession, exception: Exception): Seq[Row] = {
if (dataMapProvider != null) {
- dataMapProvider.freeMeta(mainTable, dataMapSchema)
+ dataMapProvider.cleanMeta()
}
Seq.empty
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapRefreshCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapRefreshCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapRefreshCommand.scala
index 1c39e85..4f3b7bc 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapRefreshCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapRefreshCommand.scala
@@ -21,11 +21,9 @@ import org.apache.spark.sql.{CarbonEnv, Row, SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.execution.command.DataCommand
-import org.apache.carbondata.core.datamap.DataMapStoreManager
+import org.apache.carbondata.core.datamap.{DataMapRegistry, DataMapStoreManager}
import org.apache.carbondata.core.datamap.status.DataMapStatusManager
-import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider
-import org.apache.carbondata.datamap.DataMapManager
-import org.apache.carbondata.datamap.lucene.LuceneDataMapRefreshRDD
+import org.apache.carbondata.datamap.{DataMapManager, IndexDataMapRefreshRDD}
/**
* Refresh the datamaps through sync with main table data. After sync with parent table's it enables
@@ -37,6 +35,7 @@ case class CarbonDataMapRefreshCommand(
override def processData(sparkSession: SparkSession): Seq[Row] = {
val schema = DataMapStoreManager.getInstance().getDataMapSchema(dataMapName)
+
val table = tableIdentifier match {
case Some(identifier) =>
CarbonEnv.getCarbonTable(identifier)(sparkSession)
@@ -46,13 +45,9 @@ case class CarbonDataMapRefreshCommand(
schema.getRelationIdentifier.getTableName
)(sparkSession)
}
- // Sync the datamap with parent table
- if (DataMapClassProvider.LUCENE.getShortName.endsWith(schema.getProviderName)) {
- LuceneDataMapRefreshRDD.refreshDataMap(sparkSession, table, schema)
- } else {
- val provider = DataMapManager.get().getDataMapProvider(schema, sparkSession)
- provider.rebuild(table, schema)
- }
+ val provider = DataMapManager.get().getDataMapProvider(table, schema, sparkSession)
+ provider.rebuild()
+
// After sync success enable the datamap.
DataMapStatusManager.enableDataMap(dataMapName)
Seq.empty
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
index 19e6500..cf5a4ae 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
@@ -140,9 +140,10 @@ case class CarbonDropDataMapCommand(
dbName,
tableName))(sparkSession)
if (dataMapProvider == null) {
- dataMapProvider = DataMapManager.get.getDataMapProvider(dataMapSchema, sparkSession)
+ dataMapProvider =
+ DataMapManager.get.getDataMapProvider(mainTable, dataMapSchema, sparkSession)
}
- dataMapProvider.freeMeta(mainTable, dataMapSchema)
+ dataMapProvider.cleanMeta()
// fires the event after dropping datamap from main table schema
val dropDataMapPostEvent =
@@ -181,15 +182,16 @@ case class CarbonDropDataMapCommand(
Seq.empty
}
- private def dropDataMapFromSystemFolder(sparkSession: SparkSession) = {
+ private def dropDataMapFromSystemFolder(sparkSession: SparkSession): Unit = {
try {
if (dataMapSchema == null) {
dataMapSchema = DataMapStoreManager.getInstance().getDataMapSchema(dataMapName)
}
if (dataMapSchema != null) {
- dataMapProvider = DataMapManager.get.getDataMapProvider(dataMapSchema, sparkSession)
+ dataMapProvider =
+ DataMapManager.get.getDataMapProvider(mainTable, dataMapSchema, sparkSession)
DataMapStatusManager.dropDataMap(dataMapSchema.getDataMapName)
- dataMapProvider.freeMeta(mainTable, dataMapSchema)
+ dataMapProvider.cleanMeta()
}
} catch {
case e: Exception =>
@@ -202,7 +204,7 @@ case class CarbonDropDataMapCommand(
override def processData(sparkSession: SparkSession): Seq[Row] = {
// delete the table folder
if (dataMapProvider != null) {
- dataMapProvider.freeData(mainTable, dataMapSchema)
+ dataMapProvider.cleanData()
}
Seq.empty
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
index 540fdd2..62da7ed 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
@@ -147,7 +147,7 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
/**
* The syntax of datamap creation is as follows.
- * CREATE DATAMAP IF NOT EXISTS datamapName ON TABLE tableName USING 'DataMapClassName'
+ * CREATE DATAMAP IF NOT EXISTS datamapName ON TABLE tableName USING 'DataMapProviderName'
* DMPROPERTIES('KEY'='VALUE') AS SELECT COUNT(COL1) FROM tableName
*/
protected lazy val createDataMap: Parser[LogicalPlan] =
@@ -155,10 +155,11 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
opt(ontable) ~
(USING ~> stringLit) ~ (DMPROPERTIES ~> "(" ~> repsep(loadOptions, ",") <~ ")").? ~
(AS ~> restInput).? <~ opt(";") ^^ {
- case ifnotexists ~ dmname ~ tableIdent ~ className ~ dmprops ~ query =>
+ case ifnotexists ~ dmname ~ tableIdent ~ dmProviderName ~ dmprops ~ query =>
val map = dmprops.getOrElse(List[(String, String)]()).toMap[String, String]
- CarbonCreateDataMapCommand(dmname, tableIdent, className, map, query, ifnotexists.isDefined)
+ CarbonCreateDataMapCommand(dmname, tableIdent, dmProviderName, map, query,
+ ifnotexists.isDefined)
}
protected lazy val ontable: Parser[TableIdentifier] =
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapSuite.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapSuite.scala b/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapSuite.scala
index 33de06f..a8e7b6c 100644
--- a/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapSuite.scala
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapSuite.scala
@@ -25,6 +25,8 @@ import scala.util.Random
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.BeforeAndAfterAll
+import org.apache.carbondata.core.util.CarbonProperties
+
class BloomCoarseGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
val inputFile = s"$resourcesPath/bloom_datamap_input.csv"
val normalTable = "carbon_normal"
@@ -33,12 +35,38 @@ class BloomCoarseGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
val lineNum = 500000
override protected def beforeAll(): Unit = {
+ new File(CarbonProperties.getInstance().getSystemFolderLocation).delete()
createFile(inputFile, line = lineNum, start = 0)
sql(s"DROP TABLE IF EXISTS $normalTable")
sql(s"DROP TABLE IF EXISTS $bloomDMSampleTable")
}
- test("test bloom datamap") {
+ private def checkQuery = {
+ checkAnswer(
+ sql(s"select * from $bloomDMSampleTable where id = 1"),
+ sql(s"select * from $normalTable where id = 1"))
+ checkAnswer(
+ sql(s"select * from $bloomDMSampleTable where id = 999"),
+ sql(s"select * from $normalTable where id = 999"))
+ checkAnswer(
+ sql(s"select * from $bloomDMSampleTable where city = 'city_1'"),
+ sql(s"select * from $normalTable where city = 'city_1'"))
+ checkAnswer(
+ sql(s"select * from $bloomDMSampleTable where city = 'city_999'"),
+ sql(s"select * from $normalTable where city = 'city_999'"))
+ checkAnswer(
+ sql(s"select count(distinct id), count(distinct name), count(distinct city)," +
+ s" count(distinct s1), count(distinct s2) from $bloomDMSampleTable"),
+ sql(s"select count(distinct id), count(distinct name), count(distinct city)," +
+ s" count(distinct s1), count(distinct s2) from $normalTable"))
+ checkAnswer(
+ sql(s"select min(id), max(id), min(name), max(name), min(city), max(city)" +
+ s" from $bloomDMSampleTable"),
+ sql(s"select min(id), max(id), min(name), max(name), min(city), max(city)" +
+ s" from $normalTable"))
+ }
+
+ test("test create bloom datamap on table with existing data") {
sql(
s"""
| CREATE TABLE $normalTable(id INT, name STRING, city STRING, age INT,
@@ -54,44 +82,72 @@ class BloomCoarseGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
sql(
s"""
| CREATE DATAMAP $dataMapName ON TABLE $bloomDMSampleTable
- | USING '${classOf[BloomCoarseGrainDataMapFactory].getName}'
- | DMProperties('BLOOM_COLUMNS'='city,id', 'BLOOM_SIZE'='640000')
+ | USING 'bloomfilter'
+ | DMProperties('INDEX_COLUMNS'='city,id', 'BLOOM_SIZE'='640000')
""".stripMargin)
+ // load two segments
+ (1 to 2).foreach { i =>
+ sql(
+ s"""
+ | LOAD DATA LOCAL INPATH '$inputFile' INTO TABLE $normalTable
+ | OPTIONS('header'='false')
+ """.stripMargin)
+ sql(
+ s"""
+ | LOAD DATA LOCAL INPATH '$inputFile' INTO TABLE $bloomDMSampleTable
+ | OPTIONS('header'='false')
+ """.stripMargin)
+ }
+
+ sql(s"SHOW DATAMAP ON TABLE $bloomDMSampleTable").show(false)
+ checkExistence(sql(s"SHOW DATAMAP ON TABLE $bloomDMSampleTable"), true, dataMapName)
+ checkQuery
+ sql(s"DROP TABLE IF EXISTS $normalTable")
+ sql(s"DROP TABLE IF EXISTS $bloomDMSampleTable")
+ }
+
+ test("test create bloom datamap and refresh datamap") {
sql(
s"""
- | LOAD DATA LOCAL INPATH '$inputFile' INTO TABLE $normalTable
- | OPTIONS('header'='false')
- """.stripMargin)
+ | CREATE TABLE $normalTable(id INT, name STRING, city STRING, age INT,
+ | s1 STRING, s2 STRING, s3 STRING, s4 STRING, s5 STRING, s6 STRING, s7 STRING, s8 STRING)
+ | STORED BY 'carbondata' TBLPROPERTIES('table_blocksize'='128')
+ | """.stripMargin)
sql(
s"""
- | LOAD DATA LOCAL INPATH '$inputFile' INTO TABLE $bloomDMSampleTable
- | OPTIONS('header'='false')
- """.stripMargin)
-
- sql(s"show datamap on table $bloomDMSampleTable").show(false)
- sql(s"select * from $bloomDMSampleTable where city = 'city_5'").show(false)
- sql(s"select * from $bloomDMSampleTable limit 5").show(false)
-
- checkExistence(sql(s"show datamap on table $bloomDMSampleTable"), true, dataMapName)
-// checkAnswer(sql(s"show datamap on table $bloomDMSampleTable"),
-// Row(dataMapName, classOf[BloomCoarseGrainDataMapFactory].getName, "(NA)"))
- checkAnswer(sql(s"select * from $bloomDMSampleTable where id = 1"),
- sql(s"select * from $normalTable where id = 1"))
- checkAnswer(sql(s"select * from $bloomDMSampleTable where id = 999"),
- sql(s"select * from $normalTable where id = 999"))
- checkAnswer(sql(s"select * from $bloomDMSampleTable where city = 'city_1'"),
- sql(s"select * from $normalTable where city = 'city_1'"))
- checkAnswer(sql(s"select * from $bloomDMSampleTable where city = 'city_999'"),
- sql(s"select * from $normalTable where city = 'city_999'"))
- checkAnswer(sql(s"select count(distinct id), count(distinct name), count(distinct city)," +
- s" count(distinct s1), count(distinct s2) from $bloomDMSampleTable"),
- sql(s"select count(distinct id), count(distinct name), count(distinct city)," +
- s" count(distinct s1), count(distinct s2) from $normalTable"))
- checkAnswer(sql(s"select min(id), max(id), min(name), max(name), min(city), max(city)" +
- s" from $bloomDMSampleTable"),
- sql(s"select min(id), max(id), min(name), max(name), min(city), max(city)" +
- s" from $normalTable"))
+ | CREATE TABLE $bloomDMSampleTable(id INT, name STRING, city STRING, age INT,
+ | s1 STRING, s2 STRING, s3 STRING, s4 STRING, s5 STRING, s6 STRING, s7 STRING, s8 STRING)
+ | STORED BY 'carbondata' TBLPROPERTIES('table_blocksize'='128')
+ | """.stripMargin)
+
+ // load two segments
+ (1 to 2).foreach { i =>
+ sql(
+ s"""
+ | LOAD DATA LOCAL INPATH '$inputFile' INTO TABLE $normalTable
+ | OPTIONS('header'='false')
+ """.stripMargin)
+ sql(
+ s"""
+ | LOAD DATA LOCAL INPATH '$inputFile' INTO TABLE $bloomDMSampleTable
+ | OPTIONS('header'='false')
+ """.stripMargin)
+ }
+
+ sql(
+ s"""
+ | CREATE DATAMAP $dataMapName ON TABLE $bloomDMSampleTable
+ | USING 'bloomfilter'
+ | DMProperties('INDEX_COLUMNS'='city,id', 'BLOOM_SIZE'='640000')
+ """.stripMargin)
+
+ sql(s"REFRESH DATAMAP $dataMapName ON TABLE $bloomDMSampleTable")
+ sql(s"SHOW DATAMAP ON TABLE $bloomDMSampleTable").show(false)
+ checkExistence(sql(s"SHOW DATAMAP ON TABLE $bloomDMSampleTable"), true, dataMapName)
+ checkQuery
+ sql(s"DROP TABLE IF EXISTS $normalTable")
+ sql(s"DROP TABLE IF EXISTS $bloomDMSampleTable")
}
// todo: will add more tests on bloom datamap, such as exception, delete datamap, show profiler
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterException.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterException.java b/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterException.java
new file mode 100644
index 0000000..6d062f1
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterException.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.processing.datamap;
+
+public class DataMapWriterException extends RuntimeException {
+ public DataMapWriterException(Throwable cause) {
+ super(cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterListener.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterListener.java b/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterListener.java
index a9e30d3..9c3d5d6 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterListener.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterListener.java
@@ -34,6 +34,7 @@ import org.apache.carbondata.core.datamap.dev.DataMapFactory;
import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datastore.page.ColumnPage;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.processing.store.TablePage;
/**
@@ -44,14 +45,13 @@ public class DataMapWriterListener {
private static final LogService LOG = LogServiceFactory.getLogService(
DataMapWriterListener.class.getCanonicalName());
- // list indexed column name -> list of data map writer
- private Map<List<String>, List<DataMapWriter>> registry = new ConcurrentHashMap<>();
+ // list indexed column -> list of data map writer
+ private Map<List<CarbonColumn>, List<DataMapWriter>> registry = new ConcurrentHashMap<>();
/**
* register all datamap writer for specified table and segment
*/
- public void registerAllWriter(CarbonTable carbonTable, String segmentId,
- String dataWritePath) {
+ public void registerAllWriter(CarbonTable carbonTable, String segmentId, String taskNo) {
List<TableDataMap> tableIndices;
try {
tableIndices = DataMapStoreManager.getInstance().getAllDataMap(carbonTable);
@@ -62,7 +62,7 @@ public class DataMapWriterListener {
if (tableIndices != null) {
for (TableDataMap tableDataMap : tableIndices) {
DataMapFactory factory = tableDataMap.getDataMapFactory();
- register(factory, segmentId, dataWritePath);
+ register(factory, segmentId, taskNo);
}
}
}
@@ -70,7 +70,7 @@ public class DataMapWriterListener {
/**
* Register a DataMapWriter
*/
- private void register(DataMapFactory factory, String segmentId, String dataWritePath) {
+ private void register(DataMapFactory factory, String segmentId, String taskNo) {
assert (factory != null);
assert (segmentId != null);
DataMapMeta meta = factory.getMeta();
@@ -78,9 +78,15 @@ public class DataMapWriterListener {
// if data map does not have meta, no need to register
return;
}
- List<String> columns = factory.getMeta().getIndexedColumns();
+ List<CarbonColumn> columns = factory.getMeta().getIndexedColumns();
List<DataMapWriter> writers = registry.get(columns);
- DataMapWriter writer = factory.createWriter(new Segment(segmentId, null, null), dataWritePath);
+ DataMapWriter writer = null;
+ try {
+ writer = factory.createWriter(new Segment(segmentId), taskNo);
+ } catch (IOException e) {
+ LOG.error("Failed to create DataMapWriter: " + e.getMessage());
+ throw new DataMapWriterException(e);
+ }
if (writers != null) {
writers.add(writer);
} else {
@@ -91,10 +97,10 @@ public class DataMapWriterListener {
LOG.info("DataMapWriter " + writer + " added");
}
- public void onBlockStart(String blockId, String blockPath, String taskName) throws IOException {
+ public void onBlockStart(String blockId) throws IOException {
for (List<DataMapWriter> writers : registry.values()) {
for (DataMapWriter writer : writers) {
- writer.onBlockStart(blockId, taskName);
+ writer.onBlockStart(blockId);
}
}
}
@@ -107,7 +113,7 @@ public class DataMapWriterListener {
}
}
- public void onBlockletStart(int blockletId) {
+ public void onBlockletStart(int blockletId) throws IOException {
for (List<DataMapWriter> writers : registry.values()) {
for (DataMapWriter writer : writers) {
writer.onBlockletStart(blockletId);
@@ -115,7 +121,7 @@ public class DataMapWriterListener {
}
}
- public void onBlockletEnd(int blockletId) {
+ public void onBlockletEnd(int blockletId) throws IOException {
for (List<DataMapWriter> writers : registry.values()) {
for (DataMapWriter writer : writers) {
writer.onBlockletEnd(blockletId);
@@ -130,16 +136,18 @@ public class DataMapWriterListener {
* @param tablePage page data
*/
public void onPageAdded(int blockletId, int pageId, TablePage tablePage) throws IOException {
- Set<Map.Entry<List<String>, List<DataMapWriter>>> entries = registry.entrySet();
- for (Map.Entry<List<String>, List<DataMapWriter>> entry : entries) {
- List<String> indexedColumns = entry.getKey();
+ Set<Map.Entry<List<CarbonColumn>, List<DataMapWriter>>> entries = registry.entrySet();
+ for (Map.Entry<List<CarbonColumn>, List<DataMapWriter>> entry : entries) {
+ List<CarbonColumn> indexedColumns = entry.getKey();
ColumnPage[] pages = new ColumnPage[indexedColumns.size()];
for (int i = 0; i < indexedColumns.size(); i++) {
- pages[i] = tablePage.getColumnPage(indexedColumns.get(i));
+ pages[i] = tablePage.getColumnPage(indexedColumns.get(i).getColName());
}
List<DataMapWriter> writers = entry.getValue();
+ int pageSize = pages[0].getPageSize();
+
for (DataMapWriter writer : writers) {
- writer.onPageAdded(blockletId, pageId, pages);
+ writer.onPageAdded(blockletId, pageId, pageSize, pages);
}
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/processing/src/main/java/org/apache/carbondata/processing/loading/steps/CarbonRowDataWriterProcessorStepImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/steps/CarbonRowDataWriterProcessorStepImpl.java b/processing/src/main/java/org/apache/carbondata/processing/loading/steps/CarbonRowDataWriterProcessorStepImpl.java
index ea75cd2..edf67a7 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/loading/steps/CarbonRowDataWriterProcessorStepImpl.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/loading/steps/CarbonRowDataWriterProcessorStepImpl.java
@@ -154,7 +154,7 @@ public class CarbonRowDataWriterProcessorStepImpl extends AbstractDataLoadProces
return null;
}
- private void doExecute(Iterator<CarbonRowBatch> iterator, int iteratorIndex) {
+ private void doExecute(Iterator<CarbonRowBatch> iterator, int iteratorIndex) throws IOException {
String[] storeLocation = getStoreLocation(tableIdentifier);
CarbonFactDataHandlerModel model = CarbonFactDataHandlerModel.createCarbonFactDataHandlerModel(
configuration, storeLocation, 0, iteratorIndex);
@@ -302,7 +302,12 @@ public class CarbonRowDataWriterProcessorStepImpl extends AbstractDataLoadProces
}
@Override public void run() {
- doExecute(this.iterator, iteratorIndex);
+ try {
+ doExecute(this.iterator, iteratorIndex);
+ } catch (IOException e) {
+ LOGGER.error(e);
+ throw new RuntimeException(e);
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
index 2ec85fc..a725936 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
@@ -22,7 +22,6 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Random;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.TableSpec;
@@ -233,10 +232,8 @@ public class CarbonFactDataHandlerModel {
CarbonFactDataHandlerModel carbonFactDataHandlerModel = new CarbonFactDataHandlerModel();
carbonFactDataHandlerModel.setSchemaUpdatedTimeStamp(configuration.getSchemaUpdatedTimeStamp());
- carbonFactDataHandlerModel.setDatabaseName(
- identifier.getDatabaseName());
- carbonFactDataHandlerModel
- .setTableName(identifier.getTableName());
+ carbonFactDataHandlerModel.setDatabaseName(identifier.getDatabaseName());
+ carbonFactDataHandlerModel.setTableName(identifier.getTableName());
carbonFactDataHandlerModel.setMeasureCount(measureCount);
carbonFactDataHandlerModel.setStoreLocation(storeLocation);
carbonFactDataHandlerModel.setDimLens(dimLens);
@@ -262,8 +259,14 @@ public class CarbonFactDataHandlerModel {
carbonFactDataHandlerModel.sortScope = CarbonDataProcessorUtil.getSortScope(configuration);
DataMapWriterListener listener = new DataMapWriterListener();
- listener.registerAllWriter(configuration.getTableSpec().getCarbonTable(),
- configuration.getSegmentId(), storeLocation[new Random().nextInt(storeLocation.length)]);
+ listener.registerAllWriter(
+ configuration.getTableSpec().getCarbonTable(),
+ configuration.getSegmentId(),
+ CarbonTablePath.getShardName(
+ carbonDataFileAttributes.getTaskId(),
+ bucketId,
+ 0,
+ String.valueOf(carbonDataFileAttributes.getFactTimeStamp())));
carbonFactDataHandlerModel.dataMapWriterlistener = listener;
carbonFactDataHandlerModel.writingCoresCount = configuration.getWritingCoresCount();
@@ -328,7 +331,12 @@ public class CarbonFactDataHandlerModel {
listener.registerAllWriter(
loadModel.getCarbonDataLoadSchema().getCarbonTable(),
loadModel.getSegmentId(),
- tempStoreLocation[new Random().nextInt(tempStoreLocation.length)]);
+ CarbonTablePath.getShardName(
+ CarbonTablePath.DataFileUtil.getTaskIdFromTaskNo(loadModel.getTaskNo()),
+ carbonFactDataHandlerModel.getBucketId(),
+ carbonFactDataHandlerModel.getTaskExtension(),
+ String.valueOf(loadModel.getFactTimeStamp())));
+
carbonFactDataHandlerModel.dataMapWriterlistener = listener;
return carbonFactDataHandlerModel;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java b/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
index e5c90e0..6e557cd 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
@@ -250,8 +250,7 @@ public abstract class AbstractFactDataWriter implements CarbonFactDataWriter {
private void notifyDataMapBlockStart() {
if (listener != null) {
try {
- String taskName = CarbonTablePath.getUniqueTaskName(carbonDataFileName);
- listener.onBlockStart(carbonDataFileName, carbonDataFileHdfsPath, taskName);
+ listener.onBlockStart(carbonDataFileName);
} catch (IOException e) {
throw new CarbonDataWriterException("Problem while writing datamap", e);
}
[4/4] carbondata git commit: [CARBONDATA-2415] Support Refresh
DataMap command for all Index datamap
Posted by ra...@apache.org.
[CARBONDATA-2415] Support Refresh DataMap command for all Index datamap
Refactor DataMapWriter interface to accept row instead of column page when adding data. This refactory need to modify a lot of test case files
Add REFRESH DATAMAP support for all index datamap including Lucene and Bloom
Make IndexDataMapRefreshRDD generic for all index datamap
For all index datamap, added a unified DMPROPERTIES called INDEX_COLUMNS to specify the column name list for indexing
This closes #2254
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/9db662a2
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/9db662a2
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/9db662a2
Branch: refs/heads/master
Commit: 9db662a2d352e379b42495636f27f7c10bf49f6c
Parents: f2fb068
Author: Jacky Li <ja...@qq.com>
Authored: Tue May 1 14:51:08 2018 +0800
Committer: ravipesala <ra...@gmail.com>
Committed: Sun May 6 14:20:01 2018 +0530
----------------------------------------------------------------------
.../core/constants/CarbonCommonConstants.java | 6 +
.../carbondata/core/datamap/DataMapChooser.java | 8 +-
.../carbondata/core/datamap/DataMapMeta.java | 21 +-
.../core/datamap/DataMapProvider.java | 53 ++--
.../core/datamap/DataMapRegistry.java | 44 ++-
.../core/datamap/DataMapStoreManager.java | 27 +-
.../core/datamap/IndexDataMapProvider.java | 138 --------
.../apache/carbondata/core/datamap/Segment.java | 4 +
.../carbondata/core/datamap/TableDataMap.java | 4 +-
.../core/datamap/dev/DataMapFactory.java | 76 ++++-
.../core/datamap/dev/DataMapRefresher.java | 36 +++
.../core/datamap/dev/DataMapWriter.java | 54 ++--
.../cgdatamap/CoarseGrainDataMapFactory.java | 11 +-
.../dev/expr/DataMapExprWrapperImpl.java | 2 +-
.../dev/fgdatamap/FineGrainBlocklet.java | 5 +-
.../dev/fgdatamap/FineGrainDataMapFactory.java | 12 +-
.../datamap/status/DataMapStatusManager.java | 2 +-
.../carbondata/core/indexstore/Blocklet.java | 18 +-
.../core/indexstore/ExtendedBlocklet.java | 6 +-
.../blockletindex/BlockletDataMapFactory.java | 14 +-
.../schema/datamap/DataMapClassProvider.java | 17 +-
.../core/metadata/schema/table/CarbonTable.java | 30 +-
.../metadata/schema/table/DataMapSchema.java | 21 ++
.../scanner/impl/BlockletFilterScanner.java | 3 +-
.../core/util/path/CarbonTablePath.java | 34 +-
.../datamap/bloom/BloomCoarseGrainDataMap.java | 8 +-
.../bloom/BloomCoarseGrainDataMapFactory.java | 82 ++---
.../carbondata/datamap/bloom/BloomDMModel.java | 3 +
.../datamap/bloom/BloomDataMapRefresher.java | 89 ++++++
.../datamap/bloom/BloomDataMapWriter.java | 199 ++++++------
.../datamap/examples/MinMaxDataWriter.java | 85 ++---
.../examples/MinMaxIndexDataMapFactory.java | 28 +-
.../lucene/LuceneCoarseGrainDataMapFactory.java | 13 +-
.../lucene/LuceneDataMapFactoryBase.java | 110 +++----
.../datamap/lucene/LuceneDataMapRefresher.java | 224 +++++++++++++
.../datamap/lucene/LuceneDataMapWriter.java | 204 +++++-------
.../datamap/lucene/LuceneFineGrainDataMap.java | 6 +-
.../lucene/LuceneFineGrainDataMapFactory.java | 13 +-
.../lucene/LuceneIndexRefreshBuilder.java | 220 -------------
.../hadoop/api/CarbonInputFormat.java | 4 +-
.../cluster/sdv/generated/LuceneTestCase.scala | 21 +-
.../lucene/LuceneCoarseGrainDataMapSuite.scala | 2 +-
.../lucene/LuceneFineGrainDataMapSuite.scala | 84 ++---
.../testsuite/datamap/CGDataMapTestCase.scala | 100 +++---
.../testsuite/datamap/DataMapWriterSuite.scala | 78 +++--
.../testsuite/datamap/FGDataMapTestCase.scala | 66 ++--
.../testsuite/datamap/TestDataMapStatus.scala | 70 ++--
.../TestInsertAndOtherCommandConcurrent.scala | 45 +--
.../carbondata/datamap/DataMapManager.java | 13 +-
.../datamap/IndexDataMapProvider.java | 147 +++++++++
.../datamap/PreAggregateDataMapProvider.java | 30 +-
.../datamap/TimeseriesDataMapProvider.java | 10 +-
.../datamap/IndexDataMapRefreshRDD.scala | 317 +++++++++++++++++++
.../lucene/LuceneDataMapRefreshRDD.scala | 299 -----------------
.../datamap/CarbonCreateDataMapCommand.scala | 77 +++--
.../datamap/CarbonDataMapRefreshCommand.scala | 17 +-
.../datamap/CarbonDropDataMapCommand.scala | 14 +-
.../sql/parser/CarbonSpark2SqlParser.scala | 7 +-
.../bloom/BloomCoarseGrainDataMapSuite.scala | 122 +++++--
.../datamap/DataMapWriterException.java | 24 ++
.../datamap/DataMapWriterListener.java | 42 ++-
.../CarbonRowDataWriterProcessorStepImpl.java | 9 +-
.../store/CarbonFactDataHandlerModel.java | 24 +-
.../store/writer/AbstractFactDataWriter.java | 3 +-
64 files changed, 1967 insertions(+), 1588 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
----------------------------------------------------------------------
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 648f08e..f9bf220 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
@@ -1720,6 +1720,12 @@ public final class CarbonCommonConstants {
*/
public static final String CARBON_LUCENE_COMPRESSION_MODE_DEFAULT = "speed";
+ /**
+ * DMPROPERTY for Index DataMap, like lucene, bloomfilter DataMap,
+ * to indicate a list of column name to be indexed
+ */
+ public static final String INDEX_COLUMNS = "INDEX_COLUMNS";
+
private CarbonCommonConstants() {
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/DataMapChooser.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/DataMapChooser.java b/core/src/main/java/org/apache/carbondata/core/datamap/DataMapChooser.java
index 57069b8..478254d 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/DataMapChooser.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/DataMapChooser.java
@@ -279,6 +279,10 @@ public class DataMapChooser {
return null;
}
+ /**
+ * Return true if the input datamap contains the column that needed in
+ * specified expression
+ */
private boolean contains(DataMapMeta mapMeta, List<ColumnExpression> columnExpressions,
Set<ExpressionType> expressionTypes) {
if (mapMeta.getOptimizedOperation().contains(ExpressionType.TEXT_MATCH) &&
@@ -291,8 +295,8 @@ public class DataMapChooser {
}
boolean contains = true;
for (ColumnExpression expression : columnExpressions) {
- if (!mapMeta.getIndexedColumns().contains(expression.getColumnName()) || !mapMeta
- .getOptimizedOperation().containsAll(expressionTypes)) {
+ if (!mapMeta.getIndexedColumnNames().contains(expression.getColumnName()) ||
+ !mapMeta.getOptimizedOperation().containsAll(expressionTypes)) {
contains = false;
break;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/DataMapMeta.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/DataMapMeta.java b/core/src/main/java/org/apache/carbondata/core/datamap/DataMapMeta.java
index cf51b11..adf85d8 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/DataMapMeta.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/DataMapMeta.java
@@ -21,8 +21,12 @@ import java.util.List;
import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.annotations.InterfaceStability;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.scan.filter.intf.ExpressionType;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.Transformer;
+
/**
* Metadata of the datamap, set by DataMap developer
*/
@@ -31,16 +35,17 @@ import org.apache.carbondata.core.scan.filter.intf.ExpressionType;
public class DataMapMeta {
private String dataMapName;
- private List<String> indexedColumns;
+ private List<CarbonColumn> indexedColumns;
private List<ExpressionType> optimizedOperation;
- public DataMapMeta(List<String> indexedColumns, List<ExpressionType> optimizedOperation) {
+ public DataMapMeta(List<CarbonColumn> indexedColumns,
+ List<ExpressionType> optimizedOperation) {
this.indexedColumns = indexedColumns;
this.optimizedOperation = optimizedOperation;
}
- public DataMapMeta(String dataMapName, List<String> indexedColumns,
+ public DataMapMeta(String dataMapName, List<CarbonColumn> indexedColumns,
List<ExpressionType> optimizedOperation) {
this(indexedColumns, optimizedOperation);
this.dataMapName = dataMapName;
@@ -50,10 +55,18 @@ public class DataMapMeta {
return dataMapName;
}
- public List<String> getIndexedColumns() {
+ public List<CarbonColumn> getIndexedColumns() {
return indexedColumns;
}
+ public List<String> getIndexedColumnNames() {
+ return (List<String>) CollectionUtils.collect(indexedColumns, new Transformer() {
+ @Override public Object transform(Object input) {
+ return ((CarbonColumn) input).getColName();
+ }
+ });
+ }
+
public List<ExpressionType> getOptimizedOperation() {
return optimizedOperation;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/DataMapProvider.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/DataMapProvider.java b/core/src/main/java/org/apache/carbondata/core/datamap/DataMapProvider.java
index 61dcfd1..775b912 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/DataMapProvider.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/DataMapProvider.java
@@ -21,6 +21,8 @@ import java.io.IOException;
import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
+import org.apache.carbondata.common.exceptions.sql.NoSuchDataMapException;
+import org.apache.carbondata.core.datamap.dev.DataMapFactory;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
@@ -42,47 +44,61 @@ import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
*
* <br>Currently CarbonData supports following provider:
* <ol>
- * <li> preaggregate: one type of MVDataMap that do pre-aggregate of single table </li>
- * <li> timeseries: one type of MVDataMap that do pre-aggregate based on time dimension
- * of the table </li>
- * <li> class name of {@link org.apache.carbondata.core.datamap.dev.DataMapFactory}
- * implementation: Developer can implement new type of DataMap by extending
- * {@link org.apache.carbondata.core.datamap.dev.DataMapFactory} </li>
+ * <li> preaggregate: pre-aggregate table of single table </li>
+ * <li> timeseries: pre-aggregate table based on time dimension of the table </li>
+ * <li> lucene: index backed by Apache Lucene </li>
+ * <li> bloomfilter: index backed by Bloom Filter </li>
* </ol>
*
* @since 1.4.0
*/
@InterfaceAudience.Internal
-public interface DataMapProvider {
+public abstract class DataMapProvider {
+
+ private CarbonTable mainTable;
+ private DataMapSchema dataMapSchema;
+
+ public DataMapProvider(CarbonTable mainTable, DataMapSchema dataMapSchema) {
+ this.mainTable = mainTable;
+ this.dataMapSchema = dataMapSchema;
+ }
+
+ protected final CarbonTable getMainTable() {
+ return mainTable;
+ }
+
+ protected final DataMapSchema getDataMapSchema() {
+ return dataMapSchema;
+ }
/**
* Initialize a datamap's metadata.
* This is called when user creates datamap, for example "CREATE DATAMAP dm ON TABLE mainTable"
* Implementation should initialize metadata for datamap, like creating table
*/
- void initMeta(CarbonTable mainTable, DataMapSchema dataMapSchema, String ctasSqlStatement)
- throws MalformedDataMapCommandException, IOException;
+ public abstract void initMeta(String ctasSqlStatement) throws MalformedDataMapCommandException,
+ IOException;
/**
* Initialize a datamap's data.
* This is called when user creates datamap, for example "CREATE DATAMAP dm ON TABLE mainTable"
* Implementation should initialize data for datamap, like creating data folders
*/
- void initData(CarbonTable mainTable);
+ public abstract void initData();
/**
- * Opposite operation of {@link #initMeta(CarbonTable, DataMapSchema, String)}.
+ * Opposite operation of {@link #initMeta(String)}.
* This is called when user drops datamap, for example "DROP DATAMAP dm ON TABLE mainTable"
* Implementation should clean all meta for the datamap
*/
- void freeMeta(CarbonTable mainTable, DataMapSchema dataMapSchema) throws IOException;
+ public abstract void cleanMeta() throws IOException;
/**
- * Opposite operation of {@link #initData(CarbonTable)}.
+ * Opposite operation of {@link #initData()}.
* This is called when user drops datamap, for example "DROP DATAMAP dm ON TABLE mainTable"
* Implementation should clean all data for the datamap
*/
- void freeData(CarbonTable mainTable, DataMapSchema dataMapSchema);
+ public abstract void cleanData();
/**
* Rebuild the datamap by loading all existing data from mainTable
@@ -90,19 +106,20 @@ public interface DataMapProvider {
* 1. after datamap creation and if `autoRefreshDataMap` is set to true
* 2. user manually trigger refresh datamap command
*/
- void rebuild(CarbonTable mainTable, DataMapSchema dataMapSchema) throws IOException;
+ public abstract void rebuild() throws IOException, NoSuchDataMapException;
/**
* Build the datamap incrementally by loading specified segment data
* This is called when user manually trigger refresh datamap
*/
- void incrementalBuild(CarbonTable mainTable, DataMapSchema dataMapSchema, String[] segmentIds)
- throws IOException;
+ public abstract void incrementalBuild(String[] segmentIds) throws IOException;
/**
* Provide the datamap catalog instance or null if this datamap not required to rewrite
* the query.
*/
- DataMapCatalog createDataMapCatalog();
+ public abstract DataMapCatalog createDataMapCatalog();
+
+ public abstract DataMapFactory getDataMapFactory();
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/DataMapRegistry.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/DataMapRegistry.java b/core/src/main/java/org/apache/carbondata/core/datamap/DataMapRegistry.java
index 1b6782a..8c3640a 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/DataMapRegistry.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/DataMapRegistry.java
@@ -17,12 +17,20 @@
package org.apache.carbondata.core.datamap;
+import java.lang.reflect.InvocationTargetException;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.annotations.InterfaceStability;
+import org.apache.carbondata.common.exceptions.MetadataProcessException;
+import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
+import org.apache.carbondata.core.datamap.dev.DataMap;
+import org.apache.carbondata.core.datamap.dev.DataMapFactory;
+import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
/**
* Developer can register a datamap implementation with a short name.
@@ -33,7 +41,7 @@ import org.apache.carbondata.common.annotations.InterfaceStability;
* USING 'short-name-of-the-datamap'
* }
* otherwise, user should use the class name of the datamap implementation to create the datamap
- * (subclass of {@link org.apache.carbondata.core.datamap.dev.DataMapFactory})
+ * (subclass of {@link DataMapFactory})
* <p>
* {@code
* CREATE DATAMAP dm ON TABLE table
@@ -45,14 +53,44 @@ import org.apache.carbondata.common.annotations.InterfaceStability;
public class DataMapRegistry {
private static Map<String, String> shortNameToClassName = new ConcurrentHashMap<>();
- public static void registerDataMap(String datamapClassName, String shortName) {
+ private static void registerDataMap(String datamapClassName, String shortName) {
Objects.requireNonNull(datamapClassName);
Objects.requireNonNull(shortName);
shortNameToClassName.put(shortName, datamapClassName);
}
- public static String getDataMapClassName(String shortName) {
+ private static String getDataMapClassName(String shortName) {
Objects.requireNonNull(shortName);
return shortNameToClassName.get(shortName);
}
+
+ public static DataMapFactory<? extends DataMap> getDataMapFactoryByShortName(
+ CarbonTable table, DataMapSchema dataMapSchema) throws MalformedDataMapCommandException {
+ String providerName = dataMapSchema.getProviderName();
+ try {
+ registerDataMap(
+ DataMapClassProvider.getDataMapProviderOnName(providerName).getClassName(),
+ DataMapClassProvider.getDataMapProviderOnName(providerName).getShortName());
+ } catch (UnsupportedOperationException ex) {
+ throw new MalformedDataMapCommandException("DataMap '" + providerName + "' not found", ex);
+ }
+ DataMapFactory<? extends DataMap> dataMapFactory;
+ String className = getDataMapClassName(providerName.toLowerCase());
+ if (className != null) {
+ try {
+ dataMapFactory = (DataMapFactory<? extends DataMap>)
+ Class.forName(className).getConstructors()[0].newInstance(table, dataMapSchema);
+ } catch (ClassNotFoundException ex) {
+ throw new MalformedDataMapCommandException("DataMap '" + providerName + "' not found", ex);
+ } catch (InvocationTargetException ex) {
+ throw new MalformedDataMapCommandException(ex.getTargetException().getMessage());
+ } catch (InstantiationException | IllegalAccessException | IllegalArgumentException ex) {
+ throw new MetadataProcessException(
+ "failed to create DataMap '" + providerName + "': " + ex.getMessage(), ex);
+ }
+ } else {
+ throw new MalformedDataMapCommandException("DataMap '" + providerName + "' not found");
+ }
+ return dataMapFactory;
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/DataMapStoreManager.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/DataMapStoreManager.java b/core/src/main/java/org/apache/carbondata/core/datamap/DataMapStoreManager.java
index a871d57..29a1106 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/DataMapStoreManager.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/DataMapStoreManager.java
@@ -87,7 +87,7 @@ public final class DataMapStoreManager {
List<TableDataMap> tableIndices = getAllVisibleDataMap(carbonTable);
if (tableIndices != null) {
for (TableDataMap dataMap : tableIndices) {
- if (mapType == dataMap.getDataMapFactory().getDataMapType()) {
+ if (mapType == dataMap.getDataMapFactory().getDataMapLevel()) {
dataMaps.add(dataMap);
}
}
@@ -262,22 +262,19 @@ public final class DataMapStoreManager {
}
/**
- * Return a new datamap instance for the given
- * @param dataMapSchema
- * @return
- * @throws MalformedDataMapCommandException
+ * Return a new datamap instance and registered in the store manager.
+ * The datamap is created using datamap name, datamap factory class and table identifier.
*/
- public DataMapFactory getDataMapFactoryClass(DataMapSchema dataMapSchema)
+ public DataMapFactory getDataMapFactoryClass(CarbonTable table, DataMapSchema dataMapSchema)
throws MalformedDataMapCommandException {
- DataMapFactory dataMapFactory;
try {
// try to create datamap by reflection to test whether it is a valid DataMapFactory class
- Class<? extends DataMapFactory> factoryClass =
- (Class<? extends DataMapFactory>) Class.forName(dataMapSchema.getProviderName());
- return factoryClass.newInstance();
+ return (DataMapFactory)
+ Class.forName(dataMapSchema.getProviderName()).getConstructors()[0]
+ .newInstance(table, dataMapSchema);
} catch (ClassNotFoundException e) {
// try to create DataMapClassProvider instance by taking providerName as short name
- return IndexDataMapProvider.getDataMapFactoryByShortName(dataMapSchema.getProviderName());
+ return DataMapRegistry.getDataMapFactoryByShortName(table, dataMapSchema);
} catch (Throwable e) {
throw new MetadataProcessException(
"failed to get DataMap factory for'" + dataMapSchema.getProviderName() + "'", e);
@@ -290,14 +287,13 @@ public final class DataMapStoreManager {
*/
// TODO: make it private
public TableDataMap createAndRegisterDataMap(CarbonTable table,
- DataMapSchema dataMapSchema) throws MalformedDataMapCommandException, IOException {
- DataMapFactory dataMapFactory = getDataMapFactoryClass(dataMapSchema);
+ DataMapSchema dataMapSchema) throws MalformedDataMapCommandException {
+ DataMapFactory dataMapFactory = getDataMapFactoryClass(table, dataMapSchema);
return registerDataMap(table, dataMapSchema, dataMapFactory);
}
public TableDataMap registerDataMap(CarbonTable table,
- DataMapSchema dataMapSchema, DataMapFactory dataMapFactory)
- throws IOException, MalformedDataMapCommandException {
+ DataMapSchema dataMapSchema, DataMapFactory dataMapFactory) {
String tableUniqueName = table.getCarbonTableIdentifier().getTableUniqueName();
// Just update the segmentRefreshMap with the table if not added.
getTableSegmentRefresher(table);
@@ -306,7 +302,6 @@ public final class DataMapStoreManager {
tableIndices = new ArrayList<>();
}
- dataMapFactory.init(table, dataMapSchema);
BlockletDetailsFetcher blockletDetailsFetcher;
SegmentPropertiesFetcher segmentPropertiesFetcher = null;
if (dataMapFactory instanceof BlockletDetailsFetcher) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/IndexDataMapProvider.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/IndexDataMapProvider.java b/core/src/main/java/org/apache/carbondata/core/datamap/IndexDataMapProvider.java
deleted file mode 100644
index a22bd0b..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datamap/IndexDataMapProvider.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.carbondata.core.datamap;
-
-import java.io.IOException;
-import java.util.ArrayList;
-
-import org.apache.carbondata.common.annotations.InterfaceAudience;
-import org.apache.carbondata.common.exceptions.MetadataProcessException;
-import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
-import org.apache.carbondata.core.datamap.dev.DataMapFactory;
-import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider;
-import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
-import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
-import org.apache.carbondata.core.metadata.schema.table.RelationIdentifier;
-
-@InterfaceAudience.Internal
-public class IndexDataMapProvider implements DataMapProvider {
-
- public IndexDataMapProvider() {
- }
-
- @Override
- public void initMeta(CarbonTable mainTable, DataMapSchema dataMapSchema, String ctasSqlStatement)
- throws MalformedDataMapCommandException, IOException {
- if (mainTable == null) {
- throw new MalformedDataMapCommandException(
- "Parent table is required to create index datamap");
- }
- ArrayList<RelationIdentifier> relationIdentifiers = new ArrayList<>();
- RelationIdentifier relationIdentifier =
- new RelationIdentifier(mainTable.getDatabaseName(), mainTable.getTableName(),
- mainTable.getTableInfo().getFactTable().getTableId());
- relationIdentifiers.add(relationIdentifier);
- dataMapSchema.setRelationIdentifier(relationIdentifier);
- dataMapSchema.setParentTables(relationIdentifiers);
- DataMapFactory dataMapFactory = createIndexDataMapFactory(dataMapSchema);
- DataMapStoreManager.getInstance().registerDataMap(mainTable, dataMapSchema, dataMapFactory);
- DataMapStoreManager.getInstance().saveDataMapSchema(dataMapSchema);
- }
-
- @Override
- public void initData(CarbonTable mainTable) {
- // Nothing is needed to do by default
- }
-
- @Override
- public void freeMeta(CarbonTable mainTable, DataMapSchema dataMapSchema) throws IOException {
- if (mainTable == null) {
- throw new UnsupportedOperationException("Table need to be specified in index datamaps");
- }
- DataMapStoreManager.getInstance().dropDataMapSchema(dataMapSchema.getDataMapName());
- }
-
- @Override
- public void freeData(CarbonTable mainTable, DataMapSchema dataMapSchema) {
- if (mainTable == null) {
- throw new UnsupportedOperationException("Table need to be specified in index datamaps");
- }
- DataMapStoreManager.getInstance().clearDataMap(
- mainTable.getAbsoluteTableIdentifier(), dataMapSchema.getDataMapName());
- }
-
- @Override
- public void rebuild(CarbonTable mainTable, DataMapSchema dataMapSchema) {
- // Nothing is needed to do by default
- }
-
- @Override public void incrementalBuild(CarbonTable mainTable, DataMapSchema dataMapSchema,
- String[] segmentIds) {
- throw new UnsupportedOperationException();
- }
-
- private DataMapFactory createIndexDataMapFactory(DataMapSchema dataMapSchema)
- throws MalformedDataMapCommandException {
- DataMapFactory dataMapFactory;
- try {
- // try to create DataMapClassProvider instance by taking providerName as class name
- Class<? extends DataMapFactory> providerClass =
- (Class<? extends DataMapFactory>) Class.forName(dataMapSchema.getProviderName());
- dataMapFactory = providerClass.newInstance();
- } catch (ClassNotFoundException e) {
- // try to create DataMapClassProvider instance by taking providerName as short name
- dataMapFactory = getDataMapFactoryByShortName(dataMapSchema.getProviderName());
- } catch (Throwable e) {
- throw new MetadataProcessException(
- "failed to create DataMapClassProvider '" + dataMapSchema.getProviderName() + "'", e);
- }
- return dataMapFactory;
- }
-
- public static DataMapFactory getDataMapFactoryByShortName(String providerName)
- throws MalformedDataMapCommandException {
- try {
- DataMapRegistry.registerDataMap(
- DataMapClassProvider.getDataMapProviderOnName(providerName).getClassName(),
- DataMapClassProvider.getDataMapProviderOnName(providerName).getShortName());
- } catch (UnsupportedOperationException ex) {
- throw new MalformedDataMapCommandException("DataMap '" + providerName + "' not found", ex);
- }
- DataMapFactory dataMapFactory;
- String className = DataMapRegistry.getDataMapClassName(providerName.toLowerCase());
- if (className != null) {
- try {
- Class<? extends DataMapFactory> datamapClass =
- (Class<? extends DataMapFactory>) Class.forName(className);
- dataMapFactory = datamapClass.newInstance();
- } catch (ClassNotFoundException ex) {
- throw new MalformedDataMapCommandException("DataMap '" + providerName + "' not found", ex);
- } catch (Throwable ex) {
- throw new MetadataProcessException("failed to create DataMap '" + providerName + "'", ex);
- }
- } else {
- throw new MalformedDataMapCommandException("DataMap '" + providerName + "' not found");
- }
- return dataMapFactory;
- }
-
- @Override public DataMapCatalog createDataMapCatalog() {
- // TODO create abstract class and move the default implementation there.
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/Segment.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/Segment.java b/core/src/main/java/org/apache/carbondata/core/datamap/Segment.java
index 9179bbc..476f9da 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/Segment.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/Segment.java
@@ -58,6 +58,10 @@ public class Segment implements Serializable {
*/
private LoadMetadataDetails loadMetadataDetails;
+ public Segment(String segmentNo) {
+ this.segmentNo = segmentNo;
+ }
+
/**
* ReadCommittedScope will be null. So getCommittedIndexFile will not work and will throw
* a NullPointerException. In case getCommittedIndexFile is need to be accessed then
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java b/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
index 2dc6317..314b515 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
@@ -155,13 +155,13 @@ public final class TableDataMap extends OperationEventListener {
String writePath =
identifier.getTablePath() + CarbonCommonConstants.FILE_SEPARATOR + dataMapSchema
.getDataMapName();
- if (dataMapFactory.getDataMapType() == DataMapLevel.FG) {
+ if (dataMapFactory.getDataMapLevel() == DataMapLevel.FG) {
FileFactory.mkdirs(writePath, FileFactory.getFileType(writePath));
}
for (Blocklet blocklet : blocklets) {
ExtendedBlocklet detailedBlocklet = blockletDetailsFetcher
.getExtendedBlocklet(blocklet, distributable.getSegment());
- if (dataMapFactory.getDataMapType() == DataMapLevel.FG) {
+ if (dataMapFactory.getDataMapLevel() == DataMapLevel.FG) {
String blockletwritePath =
writePath + CarbonCommonConstants.FILE_SEPARATOR + System.nanoTime();
detailedBlocklet.setDataMapWriterPath(blockletwritePath);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapFactory.java b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapFactory.java
index 20bdfb7..ae34be7 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapFactory.java
@@ -17,7 +17,9 @@
package org.apache.carbondata.core.datamap.dev;
import java.io.IOException;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
import org.apache.carbondata.core.datamap.DataMapDistributable;
@@ -27,74 +29,114 @@ import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.features.TableOperation;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.events.Event;
+import static org.apache.carbondata.core.constants.CarbonCommonConstants.INDEX_COLUMNS;
/**
- * Interface for datamap factory, it is responsible for creating the datamap.
+ * Interface for datamap of index type, it is responsible for creating the datamap.
*/
-public interface DataMapFactory<T extends DataMap> {
+public abstract class DataMapFactory<T extends DataMap> {
+
+ private CarbonTable carbonTable;
+ private DataMapSchema dataMapSchema;
+
+ public DataMapFactory(CarbonTable carbonTable, DataMapSchema dataMapSchema) {
+ this.carbonTable = carbonTable;
+ this.dataMapSchema = dataMapSchema;
+ }
+
+ public CarbonTable getCarbonTable() {
+ return carbonTable;
+ }
+
+ public DataMapSchema getDataMapSchema() {
+ return dataMapSchema;
+ }
/**
- * Initialization of Datamap factory with the carbonTable and datamap name
+ * Create a new write for this datamap, to write new data into the specified segment and shard
*/
- void init(CarbonTable carbonTable, DataMapSchema dataMapSchema)
- throws IOException, MalformedDataMapCommandException;
+ public abstract DataMapWriter createWriter(Segment segment, String shardName)
+ throws IOException;
/**
- * Return a new write for this datamap
+ * Create a new Refresher for this datamap, to rebuild the specified
+ * segment and shard data in the main table.
*/
- DataMapWriter createWriter(Segment segment, String writeDirectoryPath);
+ public abstract DataMapRefresher createRefresher(Segment segment, String shardName)
+ throws IOException;
/**
* Get the datamap for segmentid
*/
- List<T> getDataMaps(Segment segment) throws IOException;
+ public abstract List<T> getDataMaps(Segment segment) throws IOException;
/**
* Get datamaps for distributable object.
*/
- List<T> getDataMaps(DataMapDistributable distributable)
+ public abstract List<T> getDataMaps(DataMapDistributable distributable)
throws IOException;
/**
* Get all distributable objects of a segmentid
* @return
*/
- List<DataMapDistributable> toDistributable(Segment segment);
+ public abstract List<DataMapDistributable> toDistributable(Segment segment);
/**
*
* @param event
*/
- void fireEvent(Event event);
+ public abstract void fireEvent(Event event);
/**
* Clears datamap of the segment
*/
- void clear(Segment segment);
+ public abstract void clear(Segment segment);
/**
* Clear all datamaps from memory
*/
- void clear();
+ public abstract void clear();
/**
* Return metadata of this datamap
*/
- DataMapMeta getMeta();
+ public abstract DataMapMeta getMeta();
/**
* Type of datamap whether it is FG or CG
*/
- DataMapLevel getDataMapType();
+ public abstract DataMapLevel getDataMapLevel();
/**
* delete datamap data if any
*/
- void deleteDatamapData();
+ public abstract void deleteDatamapData();
/**
* This function should return true is the input operation enum will make the datamap become stale
*/
- boolean willBecomeStale(TableOperation operation);
+ public abstract boolean willBecomeStale(TableOperation operation);
+
+ /**
+ * Validate INDEX_COLUMNS property and return a array containing index column name
+ * Following will be validated
+ * 1. require INDEX_COLUMNS property
+ * 2. INDEX_COLUMNS can't contains illegal argument(empty, blank)
+ * 3. INDEX_COLUMNS can't contains duplicate same columns
+ * 4. INDEX_COLUMNS should be exists in table columns
+ */
+ public void validate() throws MalformedDataMapCommandException {
+ List<CarbonColumn> indexColumns = carbonTable.getIndexedColumns(dataMapSchema);
+ Set<String> unique = new HashSet<>();
+ for (CarbonColumn indexColumn : indexColumns) {
+ unique.add(indexColumn.getColName());
+ }
+ if (unique.size() != indexColumns.size()) {
+ throw new MalformedDataMapCommandException(INDEX_COLUMNS + " has duplicate column");
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapRefresher.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapRefresher.java b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapRefresher.java
new file mode 100644
index 0000000..770ceca
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapRefresher.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.core.datamap.dev;
+
+import java.io.IOException;
+
+import org.apache.carbondata.common.annotations.InterfaceAudience;
+
+/**
+ * Interface to rebuild the datamap for main table with existing data
+ */
+@InterfaceAudience.Developer("DataMap")
+public interface DataMapRefresher {
+ void initialize() throws IOException;
+
+ void addRow(int blockletId, int pageId, int rowId, Object[] values) throws IOException;
+
+ void finish() throws IOException;
+
+ void close() throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapWriter.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapWriter.java b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapWriter.java
index 1933f70..03a369a 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapWriter.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapWriter.java
@@ -16,14 +16,16 @@
*/
package org.apache.carbondata.core.datamap.dev;
+import java.io.File;
import java.io.IOException;
+import java.util.List;
import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.annotations.InterfaceStability;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.datastore.page.ColumnPage;
-import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;
@@ -36,17 +38,25 @@ import org.apache.carbondata.core.util.path.CarbonTablePath;
@InterfaceStability.Evolving
public abstract class DataMapWriter {
- protected AbsoluteTableIdentifier identifier;
+ protected String tablePath;
protected String segmentId;
- protected String writeDirectoryPath;
+ protected String dataMapPath;
- public DataMapWriter(AbsoluteTableIdentifier identifier, Segment segment,
- String writeDirectoryPath) {
- this.identifier = identifier;
+ private List<CarbonColumn> indexColumns;
+
+ public DataMapWriter(String tablePath, String dataMapName, List<CarbonColumn> indexColumns,
+ Segment segment, String shardName) {
+ this.tablePath = tablePath;
this.segmentId = segment.getSegmentNo();
- this.writeDirectoryPath = writeDirectoryPath;
+ this.dataMapPath = CarbonTablePath.getDataMapStorePathOnShardName(
+ tablePath, segmentId, dataMapName, shardName);
+ this.indexColumns = indexColumns;
+ }
+
+ protected final List<CarbonColumn> getIndexColumns() {
+ return indexColumns;
}
/**
@@ -54,7 +64,7 @@ public abstract class DataMapWriter {
*
* @param blockId file name of the carbondata file
*/
- public abstract void onBlockStart(String blockId, String indexShardName) throws IOException;
+ public abstract void onBlockStart(String blockId) throws IOException;
/**
* End of block notification
@@ -66,22 +76,22 @@ public abstract class DataMapWriter {
*
* @param blockletId sequence number of blocklet in the block
*/
- public abstract void onBlockletStart(int blockletId);
+ public abstract void onBlockletStart(int blockletId) throws IOException;
/**
* End of blocklet notification
*
* @param blockletId sequence number of blocklet in the block
*/
- public abstract void onBlockletEnd(int blockletId);
+ public abstract void onBlockletEnd(int blockletId) throws IOException;
/**
- * Add the column pages row to the datamap, order of pages is same as `indexColumns` in
+ * Add columnar page data to the datamap, order of field is same as `indexColumns` in
* DataMapMeta returned in DataMapFactory.
- * Implementation should copy the content of `pages` as needed, because `pages` memory
- * may be freed after this method returns, if using unsafe column page.
+ * Implementation should copy the content of it as needed, because its memory
+ * may be freed after this method returns, in case of unsafe memory
*/
- public abstract void onPageAdded(int blockletId, int pageId, ColumnPage[] pages)
+ public abstract void onPageAdded(int blockletId, int pageId, int pageSize, ColumnPage[] pages)
throws IOException;
/**
@@ -97,15 +107,15 @@ public abstract class DataMapWriter {
* @throws IOException if IO fails
*/
protected void commitFile(String dataMapFile) throws IOException {
- if (!dataMapFile.startsWith(writeDirectoryPath)) {
+ if (!dataMapFile.startsWith(dataMapPath)) {
throw new UnsupportedOperationException(
"Datamap file " + dataMapFile + " is not written in provided directory path "
- + writeDirectoryPath);
+ + dataMapPath);
}
String dataMapFileName =
- dataMapFile.substring(writeDirectoryPath.length(), dataMapFile.length());
+ dataMapFile.substring(dataMapPath.length(), dataMapFile.length());
String carbonFilePath = dataMapFileName.substring(0, dataMapFileName.lastIndexOf("/"));
- String segmentPath = CarbonTablePath.getSegmentPath(identifier.getTablePath(), segmentId);
+ String segmentPath = CarbonTablePath.getSegmentPath(tablePath, segmentId);
if (carbonFilePath.length() > 0) {
carbonFilePath = segmentPath + carbonFilePath;
FileFactory.mkdirs(carbonFilePath, FileFactory.getFileType(carbonFilePath));
@@ -115,4 +125,12 @@ public abstract class DataMapWriter {
CarbonUtil.copyCarbonDataFileToCarbonStorePath(dataMapFile, carbonFilePath, 0);
}
+
+ /**
+ * Return store path for datamap
+ */
+ public static String getDefaultDataMapPath(
+ String tablePath, String segmentId, String dataMapName) {
+ return CarbonTablePath.getSegmentPath(tablePath, segmentId) + File.separator + dataMapName;
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/dev/cgdatamap/CoarseGrainDataMapFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/dev/cgdatamap/CoarseGrainDataMapFactory.java b/core/src/main/java/org/apache/carbondata/core/datamap/dev/cgdatamap/CoarseGrainDataMapFactory.java
index f9fdafb..3de923f 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/dev/cgdatamap/CoarseGrainDataMapFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/dev/cgdatamap/CoarseGrainDataMapFactory.java
@@ -21,6 +21,8 @@ import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.annotations.InterfaceStability;
import org.apache.carbondata.core.datamap.DataMapLevel;
import org.apache.carbondata.core.datamap.dev.DataMapFactory;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
/**
* Factory for {@link CoarseGrainDataMap}
@@ -31,9 +33,14 @@ import org.apache.carbondata.core.datamap.dev.DataMapFactory;
*/
@InterfaceAudience.Developer("DataMap")
@InterfaceStability.Evolving
-public abstract class CoarseGrainDataMapFactory implements DataMapFactory<CoarseGrainDataMap> {
+public abstract class CoarseGrainDataMapFactory extends DataMapFactory<CoarseGrainDataMap> {
- @Override public DataMapLevel getDataMapType() {
+ public CoarseGrainDataMapFactory(CarbonTable carbonTable, DataMapSchema dataMapSchema) {
+ super(carbonTable, dataMapSchema);
+ }
+
+ @Override
+ public DataMapLevel getDataMapLevel() {
return DataMapLevel.CG;
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/dev/expr/DataMapExprWrapperImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/dev/expr/DataMapExprWrapperImpl.java b/core/src/main/java/org/apache/carbondata/core/datamap/dev/expr/DataMapExprWrapperImpl.java
index eff6b4d..0a3896c 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/dev/expr/DataMapExprWrapperImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/dev/expr/DataMapExprWrapperImpl.java
@@ -85,7 +85,7 @@ public class DataMapExprWrapperImpl implements DataMapExprWrapper {
}
@Override public DataMapLevel getDataMapLevel() {
- return dataMap.getDataMapFactory().getDataMapType();
+ return dataMap.getDataMapFactory().getDataMapLevel();
}
@Override public DataMapSchema getDataMapSchema() {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/dev/fgdatamap/FineGrainBlocklet.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/dev/fgdatamap/FineGrainBlocklet.java b/core/src/main/java/org/apache/carbondata/core/datamap/dev/fgdatamap/FineGrainBlocklet.java
index 9c78cc8..9d8f594 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/dev/fgdatamap/FineGrainBlocklet.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/dev/fgdatamap/FineGrainBlocklet.java
@@ -41,8 +41,9 @@ public class FineGrainBlocklet extends Blocklet implements Serializable {
private List<Page> pages;
- public FineGrainBlocklet(String taskName, String blockletId, List<Page> pages) {
- super(taskName, blockletId);
+ /** For FG, pass the shardName instead of file name */
+ public FineGrainBlocklet(String shardName, String blockletId, List<Page> pages) {
+ super(shardName, blockletId);
this.pages = pages;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/dev/fgdatamap/FineGrainDataMapFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/dev/fgdatamap/FineGrainDataMapFactory.java b/core/src/main/java/org/apache/carbondata/core/datamap/dev/fgdatamap/FineGrainDataMapFactory.java
index 0c9aaed..5e4ecb4 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/dev/fgdatamap/FineGrainDataMapFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/dev/fgdatamap/FineGrainDataMapFactory.java
@@ -20,6 +20,8 @@ import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.annotations.InterfaceStability;
import org.apache.carbondata.core.datamap.DataMapLevel;
import org.apache.carbondata.core.datamap.dev.DataMapFactory;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
/**
* Factory for {@link FineGrainDataMap}
@@ -35,10 +37,14 @@ import org.apache.carbondata.core.datamap.dev.DataMapFactory;
*/
@InterfaceAudience.Developer("DataMap")
@InterfaceStability.Evolving
-public abstract class FineGrainDataMapFactory
- implements DataMapFactory<FineGrainDataMap> {
+public abstract class FineGrainDataMapFactory extends DataMapFactory<FineGrainDataMap> {
- @Override public DataMapLevel getDataMapType() {
+ public FineGrainDataMapFactory(CarbonTable carbonTable, DataMapSchema dataMapSchema) {
+ super(carbonTable, dataMapSchema);
+ }
+
+ @Override
+ public DataMapLevel getDataMapLevel() {
return DataMapLevel.FG;
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/datamap/status/DataMapStatusManager.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/status/DataMapStatusManager.java b/core/src/main/java/org/apache/carbondata/core/datamap/status/DataMapStatusManager.java
index 31ab4e4..dcad80b 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/status/DataMapStatusManager.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/status/DataMapStatusManager.java
@@ -51,7 +51,7 @@ public class DataMapStatusManager {
return storageProvider.getDataMapStatusDetails();
}
- public static void disableDataMap(String dataMapName) throws Exception {
+ public static void disableDataMap(String dataMapName) throws IOException, NoSuchDataMapException {
DataMapSchema dataMapSchema = getDataMapSchema(dataMapName);
if (dataMapSchema != null) {
List<DataMapSchema> list = new ArrayList<>();
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/indexstore/Blocklet.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/Blocklet.java b/core/src/main/java/org/apache/carbondata/core/indexstore/Blocklet.java
index 9b40be4..777a980 100644
--- a/core/src/main/java/org/apache/carbondata/core/indexstore/Blocklet.java
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/Blocklet.java
@@ -29,13 +29,13 @@ import org.apache.carbondata.core.metadata.schema.table.Writable;
public class Blocklet implements Writable,Serializable {
/** file path of this blocklet */
- private String taskName;
+ private String filePath;
/** id to identify the blocklet inside the block (it is a sequential number) */
private String blockletId;
- public Blocklet(String taskName, String blockletId) {
- this.taskName = taskName;
+ public Blocklet(String filePath, String blockletId) {
+ this.filePath = filePath;
this.blockletId = blockletId;
}
@@ -47,17 +47,17 @@ public class Blocklet implements Writable,Serializable {
return blockletId;
}
- public String getTaskName() {
- return taskName;
+ public String getFilePath() {
+ return filePath;
}
@Override public void write(DataOutput out) throws IOException {
- out.writeUTF(taskName);
+ out.writeUTF(filePath);
out.writeUTF(blockletId);
}
@Override public void readFields(DataInput in) throws IOException {
- taskName = in.readUTF();
+ filePath = in.readUTF();
blockletId = in.readUTF();
}
@@ -67,7 +67,7 @@ public class Blocklet implements Writable,Serializable {
Blocklet blocklet = (Blocklet) o;
- if (taskName != null ? !taskName.equals(blocklet.taskName) : blocklet.taskName != null) {
+ if (filePath != null ? !filePath.equals(blocklet.filePath) : blocklet.filePath != null) {
return false;
}
return blockletId != null ?
@@ -76,7 +76,7 @@ public class Blocklet implements Writable,Serializable {
}
@Override public int hashCode() {
- int result = taskName != null ? taskName.hashCode() : 0;
+ int result = filePath != null ? filePath.hashCode() : 0;
result = 31 * result + (blockletId != null ? blockletId.hashCode() : 0);
return result;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlocklet.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlocklet.java b/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlocklet.java
index ea2752c..077b942 100644
--- a/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlocklet.java
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlocklet.java
@@ -33,8 +33,8 @@ public class ExtendedBlocklet extends Blocklet {
private String dataMapUniqueId;
- public ExtendedBlocklet(String path, String blockletId) {
- super(path, blockletId);
+ public ExtendedBlocklet(String filePath, String blockletId) {
+ super(filePath, blockletId);
}
public BlockletDetailInfo getDetailInfo() {
@@ -66,7 +66,7 @@ public class ExtendedBlocklet extends Blocklet {
}
public String getPath() {
- return getTaskName();
+ return getFilePath();
}
public String getDataMapWriterPath() {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapFactory.java b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapFactory.java
index 0d7539c..c0bc2a6 100644
--- a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMapFactory.java
@@ -29,6 +29,7 @@ import org.apache.carbondata.core.datamap.DataMapDistributable;
import org.apache.carbondata.core.datamap.DataMapMeta;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.dev.DataMap;
+import org.apache.carbondata.core.datamap.dev.DataMapRefresher;
import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap;
import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMapFactory;
@@ -72,15 +73,20 @@ public class BlockletDataMapFactory extends CoarseGrainDataMapFactory
private Cache<TableBlockIndexUniqueIdentifier, CoarseGrainDataMap> cache;
- @Override
- public void init(CarbonTable carbonTable, DataMapSchema dataMapSchema) {
+ public BlockletDataMapFactory(CarbonTable carbonTable, DataMapSchema dataMapSchema) {
+ super(carbonTable, dataMapSchema);
this.identifier = carbonTable.getAbsoluteTableIdentifier();
cache = CacheProvider.getInstance()
.createCache(CacheType.DRIVER_BLOCKLET_DATAMAP);
}
@Override
- public DataMapWriter createWriter(Segment segment, String writeDirectoryPath) {
+ public DataMapWriter createWriter(Segment segment, String shardName) {
+ throw new UnsupportedOperationException("not implemented");
+ }
+
+ @Override
+ public DataMapRefresher createRefresher(Segment segment, String shardName) {
throw new UnsupportedOperationException("not implemented");
}
@@ -147,7 +153,7 @@ public class BlockletDataMapFactory extends CoarseGrainDataMapFactory
private ExtendedBlocklet getExtendedBlocklet(List<TableBlockIndexUniqueIdentifier> identifiers,
Blocklet blocklet) throws IOException {
for (TableBlockIndexUniqueIdentifier identifier : identifiers) {
- if (identifier.getIndexFileName().startsWith(blocklet.getTaskName())) {
+ if (identifier.getIndexFileName().startsWith(blocklet.getFilePath())) {
DataMap dataMap = cache.get(identifier);
return ((BlockletDataMap) dataMap).getDetailedBlocklet(blocklet.getBlockletId());
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/DataMapClassProvider.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/DataMapClassProvider.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/DataMapClassProvider.java
index 4ab400d..d5a99e8 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/DataMapClassProvider.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/DataMapClassProvider.java
@@ -29,7 +29,8 @@ package org.apache.carbondata.core.metadata.schema.datamap;
public enum DataMapClassProvider {
PREAGGREGATE("org.apache.carbondata.core.datamap.AggregateDataMap", "preaggregate"),
TIMESERIES("org.apache.carbondata.core.datamap.TimeSeriesDataMap", "timeseries"),
- LUCENE("org.apache.carbondata.datamap.lucene.LuceneFineGrainDataMapFactory","lucene");
+ LUCENE("org.apache.carbondata.datamap.lucene.LuceneFineGrainDataMapFactory","lucene"),
+ BLOOMFILTER("org.apache.carbondata.datamap.bloom.BloomCoarseGrainDataMapFactory", "bloomfilter");
/**
* Fully qualified class name of datamap
@@ -60,18 +61,6 @@ public enum DataMapClassProvider {
dataMapClass.equalsIgnoreCase(shortName)));
}
- public static DataMapClassProvider getDataMapProvider(String dataMapClass) {
- if (TIMESERIES.isEqual(dataMapClass)) {
- return TIMESERIES;
- } else if (PREAGGREGATE.isEqual(dataMapClass)) {
- return PREAGGREGATE;
- } else if (LUCENE.isEqual(dataMapClass)) {
- return LUCENE;
- } else {
- throw new UnsupportedOperationException("Unknown datamap provider/class " + dataMapClass);
- }
- }
-
public static DataMapClassProvider getDataMapProviderOnName(String dataMapShortname) {
if (TIMESERIES.isEqual(dataMapShortname)) {
return TIMESERIES;
@@ -79,6 +68,8 @@ public enum DataMapClassProvider {
return PREAGGREGATE;
} else if (LUCENE.isEqual(dataMapShortname)) {
return LUCENE;
+ } else if (BLOOMFILTER.isEqual(dataMapShortname)) {
+ return BLOOMFILTER;
} else {
throw new UnsupportedOperationException("Unknown datamap provider" + dataMapShortname);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTable.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTable.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTable.java
index 1875237..4178d8a 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTable.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTable.java
@@ -27,6 +27,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
import org.apache.carbondata.common.logging.LogService;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
@@ -980,8 +981,11 @@ public class CarbonTable implements Serializable {
if (!datamaps.isEmpty()) {
for (TableDataMap dataMap : datamaps) {
DataMapFactory factoryClass =
- DataMapStoreManager.getInstance().getDataMapFactoryClass(dataMap.getDataMapSchema());
- return !factoryClass.willBecomeStale(operation);
+ DataMapStoreManager.getInstance().getDataMapFactoryClass(
+ carbonTable, dataMap.getDataMapSchema());
+ if (factoryClass.willBecomeStale(operation)) {
+ return false;
+ }
}
}
} catch (Exception e) {
@@ -993,4 +997,26 @@ public class CarbonTable implements Serializable {
return true;
}
+ /**
+ * Get all index columns specified by dataMapSchema
+ */
+ public List<CarbonColumn> getIndexedColumns(DataMapSchema dataMapSchema)
+ throws MalformedDataMapCommandException {
+ String[] columns = dataMapSchema.getIndexColumns();
+ List<CarbonColumn> indexColumn = new ArrayList<>(columns.length);
+ for (String column : columns) {
+ CarbonColumn carbonColumn = getColumnByName(getTableName(), column.trim().toLowerCase());
+ if (carbonColumn == null) {
+ throw new MalformedDataMapCommandException(String.format(
+ "column '%s' does not exist in table. Please check create DataMap statement.",
+ column));
+ }
+ if (carbonColumn.getColName().isEmpty()) {
+ throw new MalformedDataMapCommandException(
+ CarbonCommonConstants.INDEX_COLUMNS + " contains invalid column name");
+ }
+ indexColumn.add(carbonColumn);
+ }
+ return indexColumn;
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DataMapSchema.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DataMapSchema.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DataMapSchema.java
index 235c312..b22a3d4 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DataMapSchema.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DataMapSchema.java
@@ -26,9 +26,12 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
+import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider;
+import static org.apache.carbondata.core.constants.CarbonCommonConstants.INDEX_COLUMNS;
import com.google.gson.Gson;
+import org.apache.commons.lang.StringUtils;
/**
* It is the new schama of datamap and it has less fields compare to {{@link DataMapSchema}}
@@ -201,6 +204,24 @@ public class DataMapSchema implements Serializable, Writable {
}
}
+ /**
+ * Return the list of column name
+ */
+ public String[] getIndexColumns()
+ throws MalformedDataMapCommandException {
+ String columns = getProperties().get(INDEX_COLUMNS);
+ if (columns == null) {
+ columns = getProperties().get(INDEX_COLUMNS.toLowerCase());
+ }
+ if (columns == null) {
+ throw new MalformedDataMapCommandException(INDEX_COLUMNS + " DMPROPERTY is required");
+ } else if (StringUtils.isBlank(columns)) {
+ throw new MalformedDataMapCommandException(INDEX_COLUMNS + " DMPROPERTY is blank");
+ } else {
+ return columns.split(",", -1);
+ }
+ }
+
@Override public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/scan/scanner/impl/BlockletFilterScanner.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/scanner/impl/BlockletFilterScanner.java b/core/src/main/java/org/apache/carbondata/core/scan/scanner/impl/BlockletFilterScanner.java
index 37df0e5..57d8177 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/scanner/impl/BlockletFilterScanner.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/scanner/impl/BlockletFilterScanner.java
@@ -164,7 +164,8 @@ public class BlockletFilterScanner extends BlockletFullScanner {
totalBlockletStatistic.addCountStatistic(QueryStatisticsConstants.TOTAL_BLOCKLET_NUM,
totalBlockletStatistic.getCount() + 1);
// set the indexed data if it has any during fgdatamap pruning.
- rawBlockletColumnChunks.setBitSetGroup(rawBlockletColumnChunks.getDataBlock().getIndexedData());
+ BitSetGroup fgBitSetGroup = rawBlockletColumnChunks.getDataBlock().getIndexedData();
+ rawBlockletColumnChunks.setBitSetGroup(fgBitSetGroup);
// apply filter on actual data, for each page
BitSetGroup bitSetGroup = this.filterExecuter.applyFilter(rawBlockletColumnChunks,
useBitSetPipeLine);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java b/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
index 0538e7f..b9f4838 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
@@ -330,6 +330,11 @@ public class CarbonTablePath {
+ bucketNumber + "-" + factUpdateTimeStamp + CARBON_DATA_EXT;
}
+ public static String getShardName(Long taskNo, int bucketNumber, int batchNo,
+ String factUpdateTimeStamp) {
+ return taskNo + BATCH_PREFIX + batchNo + "-" + bucketNumber + "-" + factUpdateTimeStamp;
+ }
+
/**
* Below method will be used to get the carbon index filename
*
@@ -394,6 +399,24 @@ public class CarbonTablePath {
}
/**
+ * Return store path for datamap based on the taskNo,if three tasks get launched during loading,
+ * then three folders will be created based on the shard name and lucene index file will be
+ * written into those folders
+ *
+ * @return store path based on index shard name
+ */
+ public static String getDataMapStorePathOnShardName(String tablePath, String segmentId,
+ String dataMapName, String shardName) {
+ return new StringBuilder()
+ .append(getSegmentPath(tablePath, segmentId))
+ .append(File.separator)
+ .append(dataMapName)
+ .append(File.separator)
+ .append(shardName)
+ .toString();
+ }
+
+ /**
* To manage data file name and composition
*/
public static class DataFileUtil {
@@ -487,6 +510,13 @@ public class CarbonTablePath {
}
/**
+ * Return task id in the carbon data file name
+ */
+ public static long getTaskId(String carbonDataFileName) {
+ return Long.parseLong(getTaskNo(carbonDataFileName).split(BATCH_PREFIX)[0]);
+ }
+
+ /**
* Return the taskId part from taskNo(include taskId + batchNo)
*/
public static long getTaskIdFromTaskNo(String taskNo) {
@@ -643,7 +673,7 @@ public class CarbonTablePath {
}
public static String getCarbonIndexFileName(String actualBlockName) {
- return getUniqueTaskName(actualBlockName) + INDEX_FILE_EXT;
+ return getShardName(actualBlockName) + INDEX_FILE_EXT;
}
/**
@@ -651,7 +681,7 @@ public class CarbonTablePath {
* @param actualBlockName
* @return
*/
- public static String getUniqueTaskName(String actualBlockName) {
+ public static String getShardName(String actualBlockName) {
return DataFileUtil.getTaskNo(actualBlockName) + "-" + DataFileUtil.getBucketNo(actualBlockName)
+ "-" + DataFileUtil.getTimeStampFromFileName(actualBlockName);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMap.java
----------------------------------------------------------------------
diff --git a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMap.java b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMap.java
index 81cccf2..725d5cd 100644
--- a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMap.java
+++ b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMap.java
@@ -62,7 +62,6 @@ import org.apache.hadoop.fs.PathFilter;
public class BloomCoarseGrainDataMap extends CoarseGrainDataMap {
private static final LogService LOGGER =
LogServiceFactory.getLogService(BloomCoarseGrainDataMap.class.getName());
- private String[] indexFilePath;
private Set<String> indexedColumn;
private List<BloomDMModel> bloomIndexList;
private Multimap<String, List<BloomDMModel>> indexCol2BloomDMList;
@@ -88,18 +87,17 @@ public class BloomCoarseGrainDataMap extends CoarseGrainDataMap {
return path.getName().endsWith(BLOOM_INDEX_SUFFIX);
}
});
- indexFilePath = new String[indexFileStatus.length];
indexedColumn = new HashSet<String>();
bloomIndexList = new ArrayList<BloomDMModel>();
indexCol2BloomDMList = ArrayListMultimap.create();
for (int i = 0; i < indexFileStatus.length; i++) {
- indexFilePath[i] = indexFileStatus[i].getPath().toString();
String indexfilename = indexFileStatus[i].getPath().getName();
String indexCol =
indexfilename.substring(0, indexfilename.length() - BLOOM_INDEX_SUFFIX.length());
indexedColumn.add(indexCol);
- bloomIndexList.addAll(readBloomIndex(indexFilePath[i]));
- indexCol2BloomDMList.put(indexCol, readBloomIndex(indexFilePath[i]));
+ List<BloomDMModel> models = readBloomIndex(indexFileStatus[i].getPath().toString());
+ bloomIndexList.addAll(models);
+ indexCol2BloomDMList.put(indexCol, models);
}
LOGGER.info("find bloom index datamap for column: "
+ StringUtils.join(indexedColumn, ", "));
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFactory.java
----------------------------------------------------------------------
diff --git a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFactory.java b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFactory.java
index b76390f..2d43c40 100644
--- a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFactory.java
+++ b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFactory.java
@@ -19,10 +19,8 @@ package org.apache.carbondata.datamap.bloom;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.HashSet;
import java.util.List;
import java.util.Objects;
-import java.util.Set;
import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
@@ -34,6 +32,7 @@ import org.apache.carbondata.core.datamap.DataMapMeta;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.dev.DataMapFactory;
import org.apache.carbondata.core.datamap.dev.DataMapModel;
+import org.apache.carbondata.core.datamap.dev.DataMapRefresher;
import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
@@ -50,15 +49,14 @@ import org.apache.carbondata.events.Event;
import org.apache.commons.lang3.StringUtils;
+/**
+ * This class is for Bloom Filter for blocklet level
+ */
@InterfaceAudience.Internal
-public class BloomCoarseGrainDataMapFactory implements DataMapFactory<CoarseGrainDataMap> {
+public class BloomCoarseGrainDataMapFactory extends DataMapFactory<CoarseGrainDataMap> {
private static final LogService LOGGER = LogServiceFactory.getLogService(
BloomCoarseGrainDataMapFactory.class.getName());
/**
- * property for indexed column
- */
- private static final String BLOOM_COLUMNS = "bloom_columns";
- /**
* property for size of bloom filter
*/
private static final String BLOOM_SIZE = "bloom_size";
@@ -67,21 +65,19 @@ public class BloomCoarseGrainDataMapFactory implements DataMapFactory<CoarseGrai
* and all the indexed value is distinct.
*/
private static final int DEFAULT_BLOOM_FILTER_SIZE = 32000 * 20;
- private CarbonTable carbonTable;
private DataMapMeta dataMapMeta;
private String dataMapName;
private int bloomFilterSize;
- @Override
- public void init(CarbonTable carbonTable, DataMapSchema dataMapSchema)
- throws IOException, MalformedDataMapCommandException {
+ public BloomCoarseGrainDataMapFactory(CarbonTable carbonTable, DataMapSchema dataMapSchema)
+ throws MalformedDataMapCommandException {
+ super(carbonTable, dataMapSchema);
Objects.requireNonNull(carbonTable);
Objects.requireNonNull(dataMapSchema);
- this.carbonTable = carbonTable;
this.dataMapName = dataMapSchema.getDataMapName();
- List<String> indexedColumns = validateAndGetIndexedColumns(dataMapSchema, carbonTable);
+ List<CarbonColumn> indexedColumns = carbonTable.getIndexedColumns(dataMapSchema);
this.bloomFilterSize = validateAndGetBloomFilterSize(dataMapSchema);
List<ExpressionType> optimizedOperations = new ArrayList<ExpressionType>();
// todo: support more optimize operations
@@ -92,40 +88,6 @@ public class BloomCoarseGrainDataMapFactory implements DataMapFactory<CoarseGrai
}
/**
- * validate Lucene DataMap BLOOM_COLUMNS
- * 1. require BLOOM_COLUMNS property
- * 2. BLOOM_COLUMNS can't contains illegal argument(empty, blank)
- * 3. BLOOM_COLUMNS can't contains duplicate same columns
- * 4. BLOOM_COLUMNS should be exists in table columns
- */
- private List<String> validateAndGetIndexedColumns(DataMapSchema dmSchema,
- CarbonTable carbonTable) throws MalformedDataMapCommandException {
- String bloomColumnsStr = dmSchema.getProperties().get(BLOOM_COLUMNS);
- if (StringUtils.isBlank(bloomColumnsStr)) {
- throw new MalformedDataMapCommandException(
- String.format("Bloom coarse datamap require proper %s property", BLOOM_COLUMNS));
- }
- String[] bloomColumns = StringUtils.split(bloomColumnsStr, ",", -1);
- List<String> bloomColumnList = new ArrayList<String>(bloomColumns.length);
- Set<String> bloomColumnSet = new HashSet<String>(bloomColumns.length);
- for (String bloomCol : bloomColumns) {
- CarbonColumn column = carbonTable.getColumnByName(carbonTable.getTableName(),
- bloomCol.trim().toLowerCase());
- if (null == column) {
- throw new MalformedDataMapCommandException(
- String.format("%s: %s does not exist in table. Please check create datamap statement",
- BLOOM_COLUMNS, bloomCol));
- }
- if (!bloomColumnSet.add(column.getColName())) {
- throw new MalformedDataMapCommandException(String.format("%s has duplicate column: %s",
- BLOOM_COLUMNS, bloomCol));
- }
- bloomColumnList.add(column.getColName());
- }
- return bloomColumnList;
- }
-
- /**
* validate Lucene DataMap BLOOM_SIZE
* 1. BLOOM_SIZE property is optional, 32000 * 20 will be the default size.
* 2. BLOOM_SIZE should be an integer that greater than 0
@@ -157,21 +119,26 @@ public class BloomCoarseGrainDataMapFactory implements DataMapFactory<CoarseGrai
}
@Override
- public DataMapWriter createWriter(Segment segment, String writeDirectoryPath) {
+ public DataMapWriter createWriter(Segment segment, String shardName) throws IOException {
LOGGER.info(
String.format("Data of BloomCoarseGranDataMap %s for table %s will be written to %s",
- this.dataMapName, this.carbonTable.getTableName() , writeDirectoryPath));
- return new BloomDataMapWriter(this.carbonTable.getAbsoluteTableIdentifier(),
- this.dataMapMeta, this.bloomFilterSize, segment, writeDirectoryPath);
+ this.dataMapName, getCarbonTable().getTableName() , shardName));
+ return new BloomDataMapWriter(getCarbonTable().getTablePath(), this.dataMapName,
+ this.dataMapMeta.getIndexedColumns(), segment, shardName, this.bloomFilterSize);
+ }
+
+ @Override
+ public DataMapRefresher createRefresher(Segment segment, String shardName) throws IOException {
+ return new BloomDataMapRefresher(getCarbonTable().getTablePath(), this.dataMapName,
+ this.dataMapMeta.getIndexedColumns(), segment, shardName, this.bloomFilterSize);
}
@Override
public List<CoarseGrainDataMap> getDataMaps(Segment segment) throws IOException {
List<CoarseGrainDataMap> dataMaps = new ArrayList<CoarseGrainDataMap>(1);
try {
- String dataMapStorePath = BloomDataMapWriter.genDataMapStorePath(
- CarbonTablePath.getSegmentPath(carbonTable.getTablePath(), segment.getSegmentNo()),
- dataMapName);
+ String dataMapStorePath = DataMapWriter.getDefaultDataMapPath(
+ getCarbonTable().getTablePath(), segment.getSegmentNo(), dataMapName);
CarbonFile[] carbonFiles = FileFactory.getCarbonFile(dataMapStorePath).listFiles();
for (CarbonFile carbonFile : carbonFiles) {
BloomCoarseGrainDataMap bloomDM = new BloomCoarseGrainDataMap();
@@ -212,13 +179,14 @@ public class BloomCoarseGrainDataMapFactory implements DataMapFactory<CoarseGrai
@Override
public void deleteDatamapData() {
- SegmentStatusManager ssm = new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier());
+ SegmentStatusManager ssm =
+ new SegmentStatusManager(getCarbonTable().getAbsoluteTableIdentifier());
try {
List<Segment> validSegments = ssm.getValidAndInvalidSegments().getValidSegments();
for (Segment segment : validSegments) {
String segmentId = segment.getSegmentNo();
String datamapPath = CarbonTablePath.getSegmentPath(
- carbonTable.getAbsoluteTableIdentifier().getTablePath(), segmentId)
+ getCarbonTable().getAbsoluteTableIdentifier().getTablePath(), segmentId)
+ File.separator + dataMapName;
if (FileFactory.isFileExist(datamapPath)) {
CarbonFile file = FileFactory.getCarbonFile(datamapPath,
@@ -241,7 +209,7 @@ public class BloomCoarseGrainDataMapFactory implements DataMapFactory<CoarseGrai
}
@Override
- public DataMapLevel getDataMapType() {
+ public DataMapLevel getDataMapLevel() {
return DataMapLevel.CG;
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDMModel.java
----------------------------------------------------------------------
diff --git a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDMModel.java b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDMModel.java
index 6351199..3cf2f3b 100644
--- a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDMModel.java
+++ b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDMModel.java
@@ -22,6 +22,9 @@ import org.apache.carbondata.common.annotations.InterfaceAudience;
import com.google.common.hash.BloomFilter;
+/**
+ * This class holds a bloom filter for one blocklet
+ */
@InterfaceAudience.Internal
public class BloomDMModel implements Serializable {
private static final long serialVersionUID = 7281578747306832771L;
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDataMapRefresher.java
----------------------------------------------------------------------
diff --git a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDataMapRefresher.java b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDataMapRefresher.java
new file mode 100644
index 0000000..cb86c39
--- /dev/null
+++ b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDataMapRefresher.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.datamap.bloom;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datamap.Segment;
+import org.apache.carbondata.core.datamap.dev.DataMapRefresher;
+import org.apache.carbondata.core.metadata.datatype.DataType;
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+/**
+ * Implementation for BloomFilter DataMap to rebuild the datamap for main table with existing data
+ */
+@InterfaceAudience.Internal
+public class BloomDataMapRefresher extends BloomDataMapWriter implements DataMapRefresher {
+
+ BloomDataMapRefresher(String tablePath, String dataMapName, List<CarbonColumn> indexColumns,
+ Segment segment, String shardName, int bloomFilterSize) throws IOException {
+ super(tablePath, dataMapName, indexColumns, segment, shardName, bloomFilterSize);
+ }
+
+ @Override
+ public void initialize() throws IOException {
+ super.resetBloomFilters();
+ }
+
+ @Override
+ public void addRow(int blockletId, int pageId, int rowId, Object[] values) {
+ if (currentBlockletId != blockletId) {
+ // new blocklet started, flush bloom filter to datamap fileh
+ super.writeBloomDataMapFile();
+ currentBlockletId = blockletId;
+ }
+ // for each indexed column, add the data to bloom filter
+ List<CarbonColumn> indexColumns = getIndexColumns();
+ for (int i = 0; i < indexColumns.size(); i++) {
+ Object data = values[i];
+ DataType dataType = indexColumns.get(i).getDataType();
+ byte[] indexValue;
+ if (DataTypes.STRING == dataType) {
+ indexValue = getStringData(data);
+ } else if (DataTypes.BYTE_ARRAY == dataType) {
+ byte[] originValue = (byte[]) data;
+ // String and byte array is LV encoded, L is short type
+ indexValue = new byte[originValue.length - 2];
+ System.arraycopy(originValue, 2, indexValue, 0, originValue.length - 2);
+ } else {
+ indexValue = CarbonUtil.getValueAsBytes(dataType, data);
+ }
+ indexBloomFilters.get(i).put(indexValue);
+ }
+ }
+
+ @Override
+ public void finish() throws IOException {
+ super.finish();
+ }
+
+ @Override
+ public void close() throws IOException {
+ releaseResouce();
+ }
+
+ @Override
+ protected byte[] getStringData(Object data) {
+ return ((String) data).getBytes(CarbonCommonConstants.DEFAULT_CHARSET_CLASS);
+ }
+}
[2/4] carbondata git commit: [CARBONDATA-2415] Support Refresh
DataMap command for all Index datamap
Posted by ra...@apache.org.
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark-common-test/src/test/scala/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMapSuite.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMapSuite.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMapSuite.scala
index c157c48..17c57b6 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMapSuite.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMapSuite.scala
@@ -35,6 +35,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
val file2 = resourcesPath + "/datamap_input.csv"
override protected def beforeAll(): Unit = {
+ new File(CarbonProperties.getInstance().getSystemFolderLocation).delete()
LuceneFineGrainDataMapSuite.createFile(file2)
sql("create database if not exists lucene")
CarbonProperties.getInstance()
@@ -68,63 +69,64 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
""".stripMargin)
}
- test("validate TEXT_COLUMNS DataMap property") {
- // require TEXT_COLUMNS
+ test("validate INDEX_COLUMNS DataMap property") {
+ // require INDEX_COLUMNS
var exception = intercept[MalformedDataMapCommandException](sql(
s"""
| CREATE DATAMAP dm1 ON TABLE datamap_test
| USING 'lucene'
""".stripMargin))
- assertResult("Lucene DataMap require proper TEXT_COLUMNS property.")(exception.getMessage)
+ assert(exception.getMessage.contains("INDEX_COLUMNS DMPROPERTY is required"))
// illegal argumnet.
exception = intercept[MalformedDataMapCommandException](sql(
s"""
| CREATE DATAMAP dm1 ON TABLE datamap_test
| USING 'lucene'
- | DMProperties('text_COLUMNS'='name, ')
+ | DMProperties('INDEX_COLUMNS'='name, ')
""".stripMargin))
- assertResult("TEXT_COLUMNS contains illegal argument.")(exception.getMessage)
+ assertResult("column '' does not exist in table. Please check create DataMap statement.")(exception.getMessage)
// not exists
exception = intercept[MalformedDataMapCommandException](sql(
s"""
| CREATE DATAMAP dm1 ON TABLE datamap_test
| USING 'lucene'
- | DMProperties('text_COLUMNS'='city,school')
+ | DMProperties('INDEX_COLUMNS'='city,school')
""".stripMargin))
- assertResult("TEXT_COLUMNS: school does not exist in table. Please check create DataMap statement.")(exception.getMessage)
+ assertResult("column 'school' does not exist in table. Please check create DataMap statement.")(exception.getMessage)
// duplicate columns
exception = intercept[MalformedDataMapCommandException](sql(
s"""
| CREATE DATAMAP dm1 ON TABLE datamap_test
| USING 'lucene'
- | DMProperties('text_COLUMNS'='name,city,name')
+ | DMProperties('INDEX_COLUMNS'='name,city,name')
""".stripMargin))
- assertResult("TEXT_COLUMNS has duplicate columns :name")(exception.getMessage)
+ assertResult("INDEX_COLUMNS has duplicate column")(exception.getMessage)
// only support String DataType
exception = intercept[MalformedDataMapCommandException](sql(
s"""
| CREATE DATAMAP dm1 ON TABLE datamap_test
| USING 'lucene'
- | DMProperties('text_COLUMNS'='city,id')
+ | DMProperties('INDEX_COLUMNS'='city,id')
""".stripMargin))
- assertResult("TEXT_COLUMNS only supports String column. Unsupported column: id, DataType: INT")(exception.getMessage)
+ assertResult("Only String column is supported, column 'id' is INT type. ")(exception.getMessage)
}
test("test lucene fine grain data map") {
+ sql("drop datamap if exists dm on table datamap_test")
sql(
s"""
| CREATE DATAMAP dm ON TABLE datamap_test
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='Name , cIty')
+ | DMProperties('INDEX_COLUMNS'='Name , cIty')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test OPTIONS('header'='false')")
@@ -138,16 +140,16 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test4 OPTIONS('header'='false')")
- sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test4 OPTIONS('header'='false')")
+ //sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test4 OPTIONS('header'='false')")
sql(
s"""
| CREATE DATAMAP dm4 ON TABLE datamap_test4
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='Name , cIty')
+ | DMProperties('INDEX_COLUMNS'='Name , cIty')
""".stripMargin)
- sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test4 OPTIONS('header'='false')")
+ //sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test4 OPTIONS('header'='false')")
sql("refresh datamap dm4 ON TABLE datamap_test4")
@@ -171,7 +173,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm12 ON TABLE datamap_test1
| USING 'org.apache.carbondata.datamap.lucene.LuceneFineGrainDataMapFactory'
- | DMProperties('TEXT_COLUMNS'='Name , cIty')
+ | DMProperties('INDEX_COLUMNS'='Name , cIty')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test1 OPTIONS('header'='false')")
@@ -202,7 +204,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm122 ON TABLE datamap_test2
| USING 'org.apache.carbondata.datamap.lucene.LuceneFineGrainDataMapFactory'
- | DMProperties('TEXT_COLUMNS'='Name , cIty')
+ | DMProperties('INDEX_COLUMNS'='Name , cIty')
""".stripMargin)
sql(
@@ -215,7 +217,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm123 ON TABLE datamap_test3
| USING 'org.apache.carbondata.datamap.lucene.LuceneFineGrainDataMapFactory'
- | DMProperties('TEXT_COLUMNS'='Name , cIty')
+ | DMProperties('INDEX_COLUMNS'='Name , cIty')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test2 OPTIONS('header'='false')")
@@ -241,17 +243,16 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE datamap_test_table
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name')
+ | DMProperties('INDEX_COLUMNS'='name')
""".stripMargin)
sql(
s"""
| CREATE DATAMAP dm1 ON TABLE datamap_test_table
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name')
+ | DMProperties('INDEX_COLUMNS'='name')
""".stripMargin)
}
- assert(exception_duplicate_column.getMessage
- .contains("datamap already exists on column(s)"))
+ assertResult("column 'name' already has datamap created")(exception_duplicate_column.getMessage)
sql("drop datamap if exists dm on table datamap_test_table")
}
@@ -267,7 +268,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE datamap_test_table
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name , city')
+ | DMProperties('INDEX_COLUMNS'='name , city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test_table OPTIONS('header'='false')")
checkAnswer(sql("SELECT * FROM datamap_test_table WHERE TEXT_MATCH('name:n99*')"),
@@ -289,7 +290,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE datamap_test_table
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name , city')
+ | DMProperties('INDEX_COLUMNS'='name , city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test_table OPTIONS('header'='false')")
checkAnswer(sql(
@@ -310,7 +311,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE datamap_test_table
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name , city')
+ | DMProperties('INDEX_COLUMNS'='name , city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test_table OPTIONS('header'='false')")
checkAnswer(sql(
@@ -331,7 +332,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE datamap_test_table
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name , city')
+ | DMProperties('INDEX_COLUMNS'='name , city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test_table OPTIONS('header'='false')")
checkAnswer(sql(
@@ -355,7 +356,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE datamap_test_table
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name , city')
+ | DMProperties('INDEX_COLUMNS'='name , city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test_table OPTIONS('header'='false')")
checkAnswer(sql("SELECT * FROM datamap_test_table WHERE TEXT_MATCH('name:n10')"),
@@ -380,7 +381,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE datamap_test_table
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name , city')
+ | DMProperties('INDEX_COLUMNS'='name , city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test_table OPTIONS('header'='false')")
checkAnswer(sql("SELECT * FROM datamap_test_table WHERE TEXT_MATCH('name:n10')"),
@@ -406,7 +407,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE datamap_test_table
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name , city')
+ | DMProperties('INDEX_COLUMNS'='name,city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test_table OPTIONS('header'='false','GLOBAL_SORT_PARTITIONS'='2')")
checkAnswer(sql("SELECT * FROM datamap_test_table WHERE TEXT_MATCH('name:n10')"),
@@ -414,7 +415,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test_table OPTIONS('header'='false','GLOBAL_SORT_PARTITIONS'='2')")
checkAnswer(sql("SELECT * FROM datamap_test_table WHERE TEXT_MATCH('name:n10')"),
sql("select * from datamap_test_table where name='n10'"))
- sql("drop datamap if exists dm on table datamap_test_table")
+ sql("DROP TABLE IF EXISTS datamap_test_table")
}
test("test lucene fine grain data map with ALTER ADD and DROP Table COLUMN on Lucene DataMap") {
@@ -429,7 +430,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm2 ON TABLE datamap_test_table
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name , city')
+ | DMProperties('INDEX_COLUMNS'='name , city')
""".stripMargin)
val exception_add_column: Exception = intercept[MalformedCarbonCommandException] {
sql("alter table dm2 add columns(city1 string)")
@@ -455,7 +456,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm2 ON TABLE datamap_test_table
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name , city')
+ | DMProperties('INDEX_COLUMNS'='name , city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test_table OPTIONS('header'='false')")
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test_table OPTIONS('header'='false')")
@@ -480,7 +481,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE datamap_test_table
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name , city')
+ | DMProperties('INDEX_COLUMNS'='name , city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test_table OPTIONS('header'='false')")
//check NOT filter with TEXTMATCH term-search
@@ -504,7 +505,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm_lucene ON TABLE datamap_main
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='c')
+ | DMProperties('INDEX_COLUMNS'='c')
""".stripMargin)
sql(
"create datamap dm_pre on table datamap_main USING 'preaggregate' as select a,sum(b) " +
@@ -528,7 +529,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE source_table
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name,city')
+ | DMProperties('INDEX_COLUMNS'='name,city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE source_table OPTIONS('header'='false')")
sql(
@@ -549,7 +550,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE datamap_test
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name , city')
+ | DMProperties('INDEX_COLUMNS'='name , city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test OPTIONS('header'='false')")
@@ -563,10 +564,11 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE datamap_test
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name , city')
+ | DMProperties('INDEX_COLUMNS'='name , city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test OPTIONS('header'='false')")
+ sql("DROP TABLE IF EXISTS tabl1")
sql(
"""
| CREATE TABLE table1(id INT, name STRING, city STRING, age INT)
@@ -591,7 +593,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE main
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name , city')
+ | DMProperties('INDEX_COLUMNS'='name , city')
""".stripMargin)
val file1 = resourcesPath + "/main.csv"
@@ -633,7 +635,7 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm124 ON TABLE datamap_test7
| USING 'org.apache.carbondata.datamap.lucene.LuceneFineGrainDataMapFactory'
- | DMProperties('TEXT_COLUMNS'='Name , cIty')
+ | DMProperties('INDEX_COLUMNS'='name , city')
""".stripMargin)
val ex1 = intercept[MalformedCarbonCommandException] {
@@ -680,13 +682,13 @@ class LuceneFineGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm2 ON TABLE datamap_test5
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='city')
+ | DMProperties('INDEX_COLUMNS'='city')
""".stripMargin)
sql(
s"""
| CREATE DATAMAP dm1 ON TABLE datamap_test5
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='Name')
+ | DMProperties('INDEX_COLUMNS'='Name')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test5 OPTIONS('header'='false')")
checkAnswer(sql("SELECT * FROM datamap_test5 WHERE TEXT_MATCH('name:n10')"),
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/CGDataMapTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/CGDataMapTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/CGDataMapTestCase.scala
index 247f0ca..b441bb4 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/CGDataMapTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/CGDataMapTestCase.scala
@@ -22,17 +22,17 @@ import scala.collection.JavaConverters._
import scala.collection.mutable.ArrayBuffer
import com.sun.xml.internal.messaging.saaj.util.ByteOutputStream
+import org.apache.hadoop.fs.Path
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.BeforeAndAfterAll
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.{DataMapDistributable, DataMapMeta, Segment}
-import org.apache.carbondata.core.datamap.dev.{DataMapModel, DataMapWriter}
+import org.apache.carbondata.core.datamap.dev.{DataMapModel, DataMapRefresher, DataMapWriter}
import org.apache.carbondata.core.datamap.dev.cgdatamap.{CoarseGrainDataMap, CoarseGrainDataMapFactory}
import org.apache.carbondata.core.datastore.FileReader
import org.apache.carbondata.core.datastore.block.SegmentProperties
import org.apache.carbondata.core.datastore.compression.SnappyCompressor
-import org.apache.carbondata.core.datastore.filesystem.{CarbonFile, CarbonFileFilter}
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.datastore.page.ColumnPage
import org.apache.carbondata.core.features.TableOperation
@@ -40,7 +40,6 @@ import org.apache.carbondata.core.indexstore.{Blocklet, PartitionSpec}
import org.apache.carbondata.core.indexstore.blockletindex.BlockletDataMapDistributable
import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema, DiskBasedDMSchemaStorageProvider}
import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonMetadata}
-import org.apache.carbondata.core.readcommitter.ReadCommittedScope
import org.apache.carbondata.core.scan.expression.Expression
import org.apache.carbondata.core.scan.expression.conditional.EqualToExpression
import org.apache.carbondata.core.scan.filter.intf.ExpressionType
@@ -50,23 +49,16 @@ import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.events.Event
import org.apache.carbondata.spark.testsuite.datacompaction.CompactionSupportGlobalSortBigFileTest
-class CGDataMapFactory extends CoarseGrainDataMapFactory {
- var identifier: AbsoluteTableIdentifier = _
- var dataMapSchema: DataMapSchema = _
-
- /**
- * Initialization of Datamap factory with the identifier and datamap name
- */
- override def init(carbonTable: CarbonTable, dataMapSchema: DataMapSchema): Unit = {
- this.identifier = carbonTable.getAbsoluteTableIdentifier
- this.dataMapSchema = dataMapSchema
- }
+class CGDataMapFactory(
+ carbonTable: CarbonTable,
+ dataMapSchema: DataMapSchema) extends CoarseGrainDataMapFactory(carbonTable, dataMapSchema) {
+ var identifier: AbsoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
/**
* Return a new write for this datamap
*/
- override def createWriter(segment: Segment, dataWritePath: String): DataMapWriter = {
- new CGDataMapWriter(identifier, segment, dataWritePath, dataMapSchema)
+ override def createWriter(segment: Segment, shardName: String): DataMapWriter = {
+ new CGDataMapWriter(carbonTable, segment, shardName, dataMapSchema)
}
/**
@@ -138,7 +130,7 @@ class CGDataMapFactory extends CoarseGrainDataMapFactory {
* Return metadata of this datamap
*/
override def getMeta: DataMapMeta = {
- new DataMapMeta(dataMapSchema.getProperties.get("indexcolumns").split(",").toList.asJava,
+ new DataMapMeta(carbonTable.getIndexedColumns(dataMapSchema),
List(ExpressionType.EQUALS, ExpressionType.IN).asJava)
}
@@ -154,6 +146,11 @@ class CGDataMapFactory extends CoarseGrainDataMapFactory {
override def willBecomeStale(feature: TableOperation): Boolean = {
false
}
+
+ override def createRefresher(segment: Segment,
+ shardName: String): DataMapRefresher = {
+ ???
+ }
}
class CGDataMap extends CoarseGrainDataMap {
@@ -162,15 +159,17 @@ class CGDataMap extends CoarseGrainDataMap {
var FileReader: FileReader = _
var filePath: String = _
val compressor = new SnappyCompressor
- var taskName: String = _
+ var shardName: String = _
/**
* It is called to load the data map to memory or to initialize it.
*/
override def init(dataMapModel: DataMapModel): Unit = {
- this.filePath = dataMapModel.getFilePath
+ val indexPath = FileFactory.getPath(dataMapModel.getFilePath)
+ this.shardName = indexPath.getName
+
+ this.filePath = dataMapModel.getFilePath + "/testcg.datamap"
val carbonFile = FileFactory.getCarbonFile(filePath)
- taskName = carbonFile.getName
val size = carbonFile.getSize
FileReader = FileFactory.getFileHolder(FileFactory.getFileType(filePath))
val footerLen = FileReader.readInt(filePath, size-4)
@@ -199,7 +198,7 @@ class CGDataMap extends CoarseGrainDataMap {
}
val meta = findMeta(value(0).getBytes)
meta.map { f=>
- new Blocklet(taskName, f._1 + "")
+ new Blocklet(shardName, f._1 + "")
}.asJava
}
@@ -237,15 +236,14 @@ class CGDataMap extends CoarseGrainDataMap {
override def isScanRequired(filterExp: FilterResolverIntf): Boolean = ???
}
-class CGDataMapWriter(identifier: AbsoluteTableIdentifier,
+class CGDataMapWriter(
+ carbonTable: CarbonTable,
segment: Segment,
- dataWritePath: String,
+ shardName: String,
dataMapSchema: DataMapSchema)
- extends DataMapWriter(identifier, segment, dataWritePath) {
-
- var taskName: String = _
+ extends DataMapWriter(carbonTable.getTablePath, dataMapSchema.getDataMapName,
+ carbonTable.getIndexedColumns(dataMapSchema), segment, shardName) {
- val cgwritepath = dataWritePath + "/" + dataMapSchema.getDataMapName +"/"
val blockletList = new ArrayBuffer[Array[Byte]]()
val maxMin = new ArrayBuffer[(Int, (Array[Byte], Array[Byte]))]()
val compressor = new SnappyCompressor
@@ -255,8 +253,7 @@ class CGDataMapWriter(identifier: AbsoluteTableIdentifier,
*
* @param blockId file name of the carbondata file
*/
- override def onBlockStart(blockId: String, taskName: String): Unit = {
- this.taskName = taskName
+ override def onBlockStart(blockId: String): Unit = {
}
/**
@@ -272,7 +269,6 @@ class CGDataMapWriter(identifier: AbsoluteTableIdentifier,
* @param blockletId sequence number of blocklet in the block
*/
override def onBlockletStart(blockletId: Int): Unit = {
-
}
/**
@@ -289,7 +285,7 @@ class CGDataMapWriter(identifier: AbsoluteTableIdentifier,
}
/**
- * Add the column pages row to the datamap, order of pages is same as `indexColumns` in
+ * Add the column pages row to the datamap, order of pages is same as `index_columns` in
* DataMapMeta returned in DataMapFactory.
*
* Implementation should copy the content of `pages` as needed, because `pages` memory
@@ -297,6 +293,7 @@ class CGDataMapWriter(identifier: AbsoluteTableIdentifier,
*/
override def onPageAdded(blockletId: Int,
pageId: Int,
+ pageSize: Int,
pages: Array[ColumnPage]): Unit = {
val size = pages(0).getPageSize
val list = new ArrayBuffer[Array[Byte]]()
@@ -321,9 +318,10 @@ class CGDataMapWriter(identifier: AbsoluteTableIdentifier,
* class.
*/
override def finish(): Unit = {
- FileFactory.mkdirs(cgwritepath, FileFactory.getFileType(cgwritepath))
- var stream: DataOutputStream = FileFactory
- .getDataOutputStream(cgwritepath + "/"+taskName, FileFactory.getFileType(cgwritepath))
+ FileFactory.mkdirs(dataMapPath, FileFactory.getFileType(dataMapPath))
+ val file = dataMapPath + "/testcg.datamap"
+ val stream: DataOutputStream = FileFactory
+ .getDataOutputStream(file, FileFactory.getFileType(file))
val out = new ByteOutputStream()
val outStream = new ObjectOutputStream(out)
outStream.writeObject(maxMin)
@@ -332,7 +330,6 @@ class CGDataMapWriter(identifier: AbsoluteTableIdentifier,
stream.write(bytes)
stream.writeInt(bytes.length)
stream.close()
- commitFile(cgwritepath + "/"+taskName)
}
@@ -365,11 +362,9 @@ class CGDataMapTestCase extends QueryTest with BeforeAndAfterAll {
| STORED BY 'org.apache.carbondata.format'
| TBLPROPERTIES('SORT_COLUMNS'='city,name', 'SORT_SCOPE'='LOCAL_SORT')
""".stripMargin)
- val table = CarbonMetadata.getInstance().getCarbonTable("default_datamap_test_cg")
- // register datamap writer
sql(s"create datamap cgdatamap on table datamap_test_cg " +
s"using '${classOf[CGDataMapFactory].getName}' " +
- s"DMPROPERTIES('indexcolumns'='name')")
+ s"DMPROPERTIES('index_columns'='name')")
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test_cg OPTIONS('header'='false')")
checkAnswer(sql("select * from datamap_test_cg where name='n502670'"),
sql("select * from normal_test where name='n502670'"))
@@ -385,8 +380,8 @@ class CGDataMapTestCase extends QueryTest with BeforeAndAfterAll {
""".stripMargin)
val table = CarbonMetadata.getInstance().getCarbonTable("default_datamap_test")
// register datamap writer
- sql(s"create datamap ggdatamap1 on table datamap_test using '${classOf[CGDataMapFactory].getName}' DMPROPERTIES('indexcolumns'='name')")
- sql(s"create datamap ggdatamap2 on table datamap_test using '${classOf[CGDataMapFactory].getName}' DMPROPERTIES('indexcolumns'='city')")
+ sql(s"create datamap ggdatamap1 on table datamap_test using '${classOf[CGDataMapFactory].getName}' DMPROPERTIES('index_columns'='name')")
+ sql(s"create datamap ggdatamap2 on table datamap_test using '${classOf[CGDataMapFactory].getName}' DMPROPERTIES('index_columns'='city')")
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test OPTIONS('header'='false')")
checkAnswer(sql("select * from datamap_test where name='n502670' and city='c2670'"),
sql("select * from normal_test where name='n502670' and city='c2670'"))
@@ -404,8 +399,8 @@ class CGDataMapTestCase extends QueryTest with BeforeAndAfterAll {
| TBLPROPERTIES('SORT_COLUMNS'='city,name', 'SORT_SCOPE'='LOCAL_SORT')
""".stripMargin)
// register datamap writer
- sql(s"create datamap $dataMapName1 on table $tableName using '${classOf[CGDataMapFactory].getName}' DMPROPERTIES('indexcolumns'='name')")
- sql(s"create datamap $dataMapName2 on table $tableName using '${classOf[CGDataMapFactory].getName}' DMPROPERTIES('indexcolumns'='city')")
+ sql(s"create datamap $dataMapName1 on table $tableName using '${classOf[CGDataMapFactory].getName}' DMPROPERTIES('index_columns'='name')")
+ sql(s"create datamap $dataMapName2 on table $tableName using '${classOf[CGDataMapFactory].getName}' DMPROPERTIES('index_columns'='city')")
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE $tableName OPTIONS('header'='false')")
// make datamap1 invisible
@@ -435,7 +430,12 @@ class CGDataMapTestCase extends QueryTest with BeforeAndAfterAll {
""".stripMargin)
val dataMapProvider = classOf[CGDataMapFactory].getName
- sql(s"create datamap test_cg_datamap on table datamap_store_test using '$dataMapProvider' as select id, name from datamap_store_test")
+ sql(
+ s"""
+ |create datamap test_cg_datamap on table datamap_store_test
+ |using '$dataMapProvider'
+ |dmproperties('index_columns'='name')
+ """.stripMargin)
val loc = DiskBasedDMSchemaStorageProvider.getSchemaPath(systemFolderStoreLocation, "test_cg_datamap")
@@ -452,7 +452,12 @@ class CGDataMapTestCase extends QueryTest with BeforeAndAfterAll {
""".stripMargin)
val dataMapProvider = classOf[CGDataMapFactory].getName
- sql(s"create datamap test_cg_datamap1 on table datamap_store_test1 using '$dataMapProvider' as select id, name from datamap_store_test")
+ sql(
+ s"""
+ |create datamap test_cg_datamap1 on table datamap_store_test1
+ |using '$dataMapProvider'
+ |dmproperties('index_columns'='name')
+ """.stripMargin)
val loc = DiskBasedDMSchemaStorageProvider.getSchemaPath(systemFolderStoreLocation, "test_cg_datamap1")
@@ -473,7 +478,12 @@ class CGDataMapTestCase extends QueryTest with BeforeAndAfterAll {
""".stripMargin)
val dataMapProvider = classOf[CGDataMapFactory].getName
- sql(s"create datamap test_cg_datamap2 on table datamap_store_test2 using '$dataMapProvider' as select id, name from datamap_store_test")
+ sql(
+ s"""
+ |create datamap test_cg_datamap2 on table datamap_store_test2
+ |using '$dataMapProvider'
+ |dmproperties('index_columns'='name')
+ """.stripMargin)
val loc = DiskBasedDMSchemaStorageProvider.getSchemaPath(systemFolderStoreLocation,"test_cg_datamap2")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala
index ec72ffb..5e0c10a 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala
@@ -21,34 +21,28 @@ import java.util
import scala.collection.JavaConverters._
-import org.apache.spark.sql.test.util.QueryTest
import org.apache.spark.sql.{DataFrame, SaveMode}
+import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.BeforeAndAfterAll
import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.datamap.{DataMapDistributable, DataMapMeta, DataMapStoreManager, Segment}
-import org.apache.carbondata.core.datamap.dev.DataMapWriter
+import org.apache.carbondata.core.datamap.{DataMapDistributable, DataMapMeta, Segment}
+import org.apache.carbondata.core.datamap.dev.{DataMapRefresher, DataMapWriter}
import org.apache.carbondata.core.datamap.dev.cgdatamap.{CoarseGrainDataMap, CoarseGrainDataMapFactory}
-import org.apache.carbondata.core.datamap.{DataMapDistributable, DataMapMeta}
import org.apache.carbondata.core.datastore.page.ColumnPage
import org.apache.carbondata.core.features.TableOperation
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.metadata.datatype.DataTypes
-import org.apache.carbondata.core.metadata.schema.table.DataMapSchema
-import org.apache.carbondata.core.readcommitter.ReadCommittedScope
import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}
import org.apache.carbondata.core.scan.filter.intf.ExpressionType
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.events.Event
-class C2DataMapFactory() extends CoarseGrainDataMapFactory {
+class C2DataMapFactory(
+ carbonTable: CarbonTable,
+ dataMapSchema: DataMapSchema) extends CoarseGrainDataMapFactory(carbonTable, dataMapSchema) {
- var identifier: AbsoluteTableIdentifier = _
-
- override def init(carbonTable: CarbonTable,
- dataMapSchema: DataMapSchema): Unit = {
- this.identifier = carbonTable.getAbsoluteTableIdentifier
- }
+ var identifier: AbsoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
override def fireEvent(event: Event): Unit = ???
@@ -60,10 +54,11 @@ class C2DataMapFactory() extends CoarseGrainDataMapFactory {
override def getDataMaps(segment: Segment): util.List[CoarseGrainDataMap] = ???
- override def createWriter(segment: Segment, dataWritePath: String): DataMapWriter =
- DataMapWriterSuite.dataMapWriterC2Mock(identifier, segment, dataWritePath)
+ override def createWriter(segment: Segment, shardName: String): DataMapWriter =
+ DataMapWriterSuite.dataMapWriterC2Mock(identifier, "testdm", segment, shardName)
- override def getMeta: DataMapMeta = new DataMapMeta(List("c2").asJava, List(ExpressionType.EQUALS).asJava)
+ override def getMeta: DataMapMeta =
+ new DataMapMeta(carbonTable.getIndexedColumns(dataMapSchema), List(ExpressionType.EQUALS).asJava)
/**
* Get all distributable objects of a segmentid
@@ -87,6 +82,11 @@ class C2DataMapFactory() extends CoarseGrainDataMapFactory {
override def willBecomeStale(operation: TableOperation): Boolean = {
false
}
+
+ override def createRefresher(segment: Segment,
+ shardName: String): DataMapRefresher = {
+ ???
+ }
}
class DataMapWriterSuite extends QueryTest with BeforeAndAfterAll {
@@ -109,7 +109,12 @@ class DataMapWriterSuite extends QueryTest with BeforeAndAfterAll {
test("test write datamap 2 pages") {
sql(s"CREATE TABLE carbon1(c1 STRING, c2 STRING, c3 INT) STORED BY 'org.apache.carbondata.format'")
// register datamap writer
- sql(s"CREATE DATAMAP test1 ON TABLE carbon1 USING '${classOf[C2DataMapFactory].getName}'")
+ sql(
+ s"""
+ | CREATE DATAMAP test1 ON TABLE carbon1
+ | USING '${classOf[C2DataMapFactory].getName}'
+ | DMPROPERTIES('index_columns'='c2')
+ """.stripMargin)
val df = buildTestData(33000)
// save dataframe to carbon file
@@ -135,8 +140,12 @@ class DataMapWriterSuite extends QueryTest with BeforeAndAfterAll {
test("test write datamap 2 blocklet") {
sql(s"CREATE TABLE carbon2(c1 STRING, c2 STRING, c3 INT) STORED BY 'org.apache.carbondata.format'")
- sql(s"CREATE DATAMAP test2 ON TABLE carbon2 USING '${classOf[C2DataMapFactory].getName}'")
-
+ sql(
+ s"""
+ | CREATE DATAMAP test2 ON TABLE carbon2
+ | USING '${classOf[C2DataMapFactory].getName}'
+ | DMPROPERTIES('index_columns'='c2')
+ """.stripMargin)
CarbonProperties.getInstance()
.addProperty("carbon.blockletgroup.size.in.mb", "1")
CarbonProperties.getInstance()
@@ -186,20 +195,21 @@ object DataMapWriterSuite {
var callbackSeq: Seq[String] = Seq[String]()
- def dataMapWriterC2Mock(identifier: AbsoluteTableIdentifier, segment: Segment,
- dataWritePath: String) =
- new DataMapWriter(identifier, segment, dataWritePath) {
-
- override def onPageAdded(
- blockletId: Int,
- pageId: Int,
- pages: Array[ColumnPage]): Unit = {
- assert(pages.length == 1)
- assert(pages(0).getDataType == DataTypes.STRING)
- val bytes: Array[Byte] = pages(0).getByteArrayPage()(0)
- assert(bytes.sameElements(Seq(0, 1, 'b'.toByte)))
- callbackSeq :+= s"add page data: blocklet $blockletId, page $pageId"
- }
+ def dataMapWriterC2Mock(identifier: AbsoluteTableIdentifier, dataMapName:String, segment: Segment,
+ shardName: String) =
+ new DataMapWriter(identifier.getTablePath, dataMapName, Seq().asJava, segment, shardName) {
+
+ override def onPageAdded(
+ blockletId: Int,
+ pageId: Int,
+ pageSize: Int,
+ pages: Array[ColumnPage]): Unit = {
+ assert(pages.length == 1)
+ assert(pages(0).getDataType == DataTypes.STRING)
+ val bytes: Array[Byte] = pages(0).getByteArrayPage()(0)
+ assert(bytes.sameElements(Seq(0, 1, 'b'.toByte)))
+ callbackSeq :+= s"add page data: blocklet $blockletId, page $pageId"
+ }
override def onBlockletEnd(blockletId: Int): Unit = {
callbackSeq :+= s"blocklet end: $blockletId"
@@ -218,7 +228,7 @@ object DataMapWriterSuite {
*
* @param blockId file name of the carbondata file
*/
- override def onBlockStart(blockId: String, taskId: String) = {
+ override def onBlockStart(blockId: String): Unit = {
callbackSeq :+= s"block start $blockId"
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/FGDataMapTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/FGDataMapTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/FGDataMapTestCase.scala
index 0422b24..976e580 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/FGDataMapTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/FGDataMapTestCase.scala
@@ -25,23 +25,20 @@ import com.sun.xml.internal.messaging.saaj.util.ByteOutputStream
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.BeforeAndAfterAll
-import org.apache.carbondata.core.datamap.dev.DataMapModel
-import org.apache.carbondata.core.datamap.{DataMapDistributable, DataMapMeta, DataMapStoreManager, Segment}
-import org.apache.carbondata.core.datamap.dev.fgdatamap.{FineGrainBlocklet, FineGrainDataMap, FineGrainDataMapFactory}
-import org.apache.carbondata.core.datamap.dev.{DataMapModel, DataMapWriter}
import org.apache.carbondata.core.datamap.{DataMapDistributable, DataMapMeta}
+import org.apache.carbondata.core.datamap.Segment
+import org.apache.carbondata.core.datamap.dev.{DataMapModel, DataMapRefresher, DataMapWriter}
+import org.apache.carbondata.core.datamap.dev.fgdatamap.{FineGrainBlocklet, FineGrainDataMap, FineGrainDataMapFactory}
import org.apache.carbondata.core.datastore.FileReader
import org.apache.carbondata.core.datastore.block.SegmentProperties
import org.apache.carbondata.core.datastore.compression.SnappyCompressor
-import org.apache.carbondata.core.datastore.filesystem.{CarbonFile, CarbonFileFilter}
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.datastore.page.ColumnPage
import org.apache.carbondata.core.features.TableOperation
-import org.apache.carbondata.core.indexstore.{Blocklet, PartitionSpec}
+import org.apache.carbondata.core.indexstore.PartitionSpec
import org.apache.carbondata.core.indexstore.blockletindex.BlockletDataMapDistributable
import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}
-import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonMetadata}
-import org.apache.carbondata.core.readcommitter.ReadCommittedScope
+import org.apache.carbondata.core.metadata.CarbonMetadata
import org.apache.carbondata.core.scan.expression.Expression
import org.apache.carbondata.core.scan.expression.conditional.EqualToExpression
import org.apache.carbondata.core.scan.filter.intf.ExpressionType
@@ -51,30 +48,21 @@ import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.events.Event
import org.apache.carbondata.spark.testsuite.datacompaction.CompactionSupportGlobalSortBigFileTest
-class FGDataMapFactory extends FineGrainDataMapFactory {
- var identifier: AbsoluteTableIdentifier = _
- var dataMapSchema: DataMapSchema = _
-
- /**
- * Initialization of Datamap factory with the identifier and datamap name
- */
- override def init(carbonTable: CarbonTable, dataMapSchema: DataMapSchema): Unit = {
- this.identifier = carbonTable.getAbsoluteTableIdentifier
- this.dataMapSchema = dataMapSchema
- }
+class FGDataMapFactory(carbonTable: CarbonTable,
+ dataMapSchema: DataMapSchema) extends FineGrainDataMapFactory(carbonTable, dataMapSchema) {
/**
* Return a new write for this datamap
*/
override def createWriter(segment: Segment, dataWritePath: String): DataMapWriter = {
- new FGDataMapWriter(identifier, segment, dataWritePath, dataMapSchema)
+ new FGDataMapWriter(carbonTable, segment, dataWritePath, dataMapSchema)
}
/**
* Get the datamap for segmentid
*/
override def getDataMaps(segment: Segment): java.util.List[FineGrainDataMap] = {
- val path = CarbonTablePath.getSegmentPath(identifier.getTablePath, segment.getSegmentNo)
+ val path = CarbonTablePath.getSegmentPath(carbonTable.getTablePath, segment.getSegmentNo)
val file = FileFactory.getCarbonFile(path+ "/" +dataMapSchema.getDataMapName)
val files = file.listFiles()
@@ -101,7 +89,7 @@ class FGDataMapFactory extends FineGrainDataMapFactory {
* @return
*/
override def toDistributable(segment: Segment): java.util.List[DataMapDistributable] = {
- val path = CarbonTablePath.getSegmentPath(identifier.getTablePath, segment.getSegmentNo)
+ val path = CarbonTablePath.getSegmentPath(carbonTable.getTablePath, segment.getSegmentNo)
val file = FileFactory.getCarbonFile(path+ "/" +dataMapSchema.getDataMapName)
val files = file.listFiles()
@@ -135,7 +123,7 @@ class FGDataMapFactory extends FineGrainDataMapFactory {
* Return metadata of this datamap
*/
override def getMeta: DataMapMeta = {
- new DataMapMeta(dataMapSchema.getProperties.get("indexcolumns").split(",").toList.asJava,
+ new DataMapMeta(carbonTable.getIndexedColumns(dataMapSchema),
List(ExpressionType.EQUALS, ExpressionType.IN).asJava)
}
@@ -152,6 +140,11 @@ class FGDataMapFactory extends FineGrainDataMapFactory {
override def willBecomeStale(operation: TableOperation): Boolean = {
false
}
+
+ override def createRefresher(segment: Segment,
+ shardName: String): DataMapRefresher = {
+ ???
+ }
}
class FGDataMap extends FineGrainDataMap {
@@ -266,12 +259,13 @@ class FGDataMap extends FineGrainDataMap {
override def isScanRequired(filterExp: FilterResolverIntf): Boolean = ???
}
-class FGDataMapWriter(identifier: AbsoluteTableIdentifier,
- segment: Segment, dataWriterPath: String, dataMapSchema: DataMapSchema)
- extends DataMapWriter(identifier, segment, dataWriterPath) {
+class FGDataMapWriter(carbonTable: CarbonTable,
+ segment: Segment, shardName: String, dataMapSchema: DataMapSchema)
+ extends DataMapWriter(carbonTable.getTablePath, dataMapSchema.getDataMapName,
+ carbonTable.getIndexedColumns(dataMapSchema), segment, shardName) {
var taskName: String = _
- val fgwritepath = dataWriterPath + "/" + dataMapSchema.getDataMapName +"/"
+ val fgwritepath = dataMapPath
var stream: DataOutputStream = _
val blockletList = new ArrayBuffer[(Array[Byte], Seq[Int], Seq[Int])]()
val maxMin = new ArrayBuffer[(Int, (Array[Byte], Array[Byte]), Long, Int)]()
@@ -283,12 +277,13 @@ class FGDataMapWriter(identifier: AbsoluteTableIdentifier,
*
* @param blockId file name of the carbondata file
*/
- override def onBlockStart(blockId: String, taskId: String): Unit = {
- this.taskName = taskId
+ override def onBlockStart(blockId: String): Unit = {
+ this.taskName = shardName
if (stream == null) {
- FileFactory.mkdirs(fgwritepath, FileFactory.getFileType(fgwritepath))
+ val path = fgwritepath.substring(0, fgwritepath.lastIndexOf("/"))
+ FileFactory.mkdirs(path, FileFactory.getFileType(path))
stream = FileFactory
- .getDataOutputStream(fgwritepath + "/"+taskName, FileFactory.getFileType(fgwritepath))
+ .getDataOutputStream(fgwritepath, FileFactory.getFileType(fgwritepath))
}
}
@@ -361,6 +356,7 @@ class FGDataMapWriter(identifier: AbsoluteTableIdentifier,
*/
override def onPageAdded(blockletId: Int,
pageId: Int,
+ pageSize: Int,
pages: Array[ColumnPage]): Unit = {
val size = pages(0).getPageSize
val list = new ArrayBuffer[(Array[Byte], Int)]()
@@ -413,7 +409,7 @@ class FGDataMapWriter(identifier: AbsoluteTableIdentifier,
stream.write(bytes)
stream.writeInt(bytes.length)
stream.close()
- commitFile(fgwritepath + "/"+taskName)
+// commitFile(fgwritepath)
}
}
@@ -449,7 +445,7 @@ class FGDataMapTestCase extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP ggdatamap ON TABLE datamap_test
| USING '${classOf[FGDataMapFactory].getName}'
- | DMPROPERTIES('indexcolumns'='name')
+ | DMPROPERTIES('index_columns'='name')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test OPTIONS('header'='false')")
checkAnswer(sql("select * from datamap_test where name='n502670'"),
@@ -470,13 +466,13 @@ class FGDataMapTestCase extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP ggdatamap1 ON TABLE datamap_test
| USING '${classOf[FGDataMapFactory].getName}'
- | DMPROPERTIES('indexcolumns'='name')
+ | DMPROPERTIES('index_columns'='name')
""".stripMargin)
sql(
s"""
| CREATE DATAMAP ggdatamap2 ON TABLE datamap_test
| USING '${classOf[FGDataMapFactory].getName}'
- | DMPROPERTIES('indexcolumns'='city')
+ | DMPROPERTIES('index_columns'='city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test OPTIONS('header'='false')")
checkAnswer(sql("select * from datamap_test where name='n502670' and city='c2670'"),
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapStatus.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapStatus.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapStatus.scala
index 037ba1e..67effda 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapStatus.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapStatus.scala
@@ -26,15 +26,14 @@ import org.scalatest.BeforeAndAfterAll
import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException
import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.datamap.dev.DataMapWriter
+import org.apache.carbondata.core.datamap.dev.{DataMapRefresher, DataMapWriter}
import org.apache.carbondata.core.datamap.dev.cgdatamap.{CoarseGrainDataMap, CoarseGrainDataMapFactory}
import org.apache.carbondata.core.datamap.status.{DataMapStatus, DataMapStatusManager}
import org.apache.carbondata.core.datamap.{DataMapDistributable, DataMapMeta, Segment}
import org.apache.carbondata.core.datastore.page.ColumnPage
import org.apache.carbondata.core.features.TableOperation
+import org.apache.carbondata.core.datastore.row.CarbonRow
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
-import org.apache.carbondata.core.metadata.schema.table.DataMapSchema
-import org.apache.carbondata.core.readcommitter.ReadCommittedScope
import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}
import org.apache.carbondata.core.scan.filter.intf.ExpressionType
import org.apache.carbondata.core.util.CarbonProperties
@@ -56,7 +55,10 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
| STORED BY 'org.apache.carbondata.format'
""".stripMargin)
sql(
- s"""create datamap statusdatamap on table datamapstatustest using '${classOf[TestDataMap].getName}' as select id,sum(age) from datamapstatustest group by id""".stripMargin)
+ s"""create datamap statusdatamap on table datamapstatustest
+ |using '${classOf[TestDataMapFactory].getName}'
+ |dmproperties('index_columns'='name')
+ | """.stripMargin)
val details = DataMapStatusManager.readDataMapStatusDetails()
@@ -74,7 +76,10 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
| STORED BY 'org.apache.carbondata.format'
""".stripMargin)
sql(
- s"""create datamap statusdatamap1 on table datamapstatustest1 using '${classOf[TestDataMap].getName}' as select id,sum(age) from datamapstatustest1 group by id""".stripMargin)
+ s"""create datamap statusdatamap1 on table datamapstatustest1
+ |using '${classOf[TestDataMapFactory].getName}'
+ |dmproperties('index_columns'='name')
+ | """.stripMargin)
var details = DataMapStatusManager.readDataMapStatusDetails()
@@ -89,7 +94,8 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
sql("DROP TABLE IF EXISTS datamapstatustest1")
}
- test("datamap status with refresh datamap") {
+ // enable it in PR2255
+ ignore("datamap status with refresh datamap") {
sql("DROP TABLE IF EXISTS datamapstatustest2")
sql(
"""
@@ -97,7 +103,10 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
| STORED BY 'org.apache.carbondata.format'
""".stripMargin)
sql(
- s"""create datamap statusdatamap2 on table datamapstatustest2 using '${classOf[TestDataMap].getName}' as select id,sum(age) from datamapstatustest1 group by id""".stripMargin)
+ s"""create datamap statusdatamap2 on table datamapstatustest2
+ |using '${classOf[TestDataMapFactory].getName}'
+ |dmproperties('index_columns'='name')
+ | """.stripMargin)
var details = DataMapStatusManager.readDataMapStatusDetails()
@@ -119,7 +128,8 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
sql("DROP TABLE IF EXISTS datamapstatustest2")
}
- test("datamap create without on table test") {
+ // enable it in PR2255
+ ignore("datamap create without on table test") {
sql("DROP TABLE IF EXISTS datamapstatustest3")
sql(
"""
@@ -128,17 +138,18 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
""".stripMargin)
intercept[MalformedDataMapCommandException] {
sql(
- s"""create datamap statusdatamap3 using '${
- classOf[TestDataMap]
- .getName
- }' as select id,sum(age) from datamapstatustest3 group by id""".stripMargin)
+ s"""create datamap statusdatamap3
+ |using '${classOf[TestDataMapFactory].getName}'
+ |dmproperties('index_columns'='name')
+ | """.stripMargin)
+
}
sql(
- s"""create datamap statusdatamap3 on table datamapstatustest3 using '${
- classOf[TestDataMap]
- .getName
- }' as select id,sum(age) from datamapstatustest3 group by id""".stripMargin)
+ s"""create datamap statusdatamap3 on table datamapstatustest3
+ |using '${classOf[TestDataMapFactory].getName}'
+ |dmproperties('index_columns'='name')
+ | """.stripMargin)
var details = DataMapStatusManager.readDataMapStatusDetails()
@@ -176,9 +187,9 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
}
}
-class TestDataMap() extends CoarseGrainDataMapFactory {
-
- private var identifier: AbsoluteTableIdentifier = _
+class TestDataMapFactory(
+ carbonTable: CarbonTable,
+ dataMapSchema: DataMapSchema) extends CoarseGrainDataMapFactory(carbonTable, dataMapSchema) {
override def fireEvent(event: Event): Unit = ???
@@ -194,9 +205,10 @@ class TestDataMap() extends CoarseGrainDataMapFactory {
???
}
- override def createWriter(segment: Segment, writeDirectoryPath: String): DataMapWriter = {
- new DataMapWriter(identifier, segment, writeDirectoryPath) {
- override def onPageAdded(blockletId: Int, pageId: Int, pages: Array[ColumnPage]): Unit = { }
+ override def createWriter(segment: Segment, shardName: String): DataMapWriter = {
+ new DataMapWriter(carbonTable.getTablePath, "testdm", carbonTable.getIndexedColumns(dataMapSchema),
+ segment, shardName) {
+ override def onPageAdded(blockletId: Int, pageId: Int, pageSize: Int, pages: Array[ColumnPage]): Unit = { }
override def onBlockletEnd(blockletId: Int): Unit = { }
@@ -204,7 +216,7 @@ class TestDataMap() extends CoarseGrainDataMapFactory {
override def onBlockletStart(blockletId: Int): Unit = { }
- override def onBlockStart(blockId: String, taskId: String): Unit = {
+ override def onBlockStart(blockId: String): Unit = {
// trigger the second SQL to execute
}
@@ -214,14 +226,11 @@ class TestDataMap() extends CoarseGrainDataMapFactory {
}
}
- override def getMeta: DataMapMeta = new DataMapMeta(List("id").asJava, Seq(ExpressionType.EQUALS).asJava)
+ override def getMeta: DataMapMeta = new DataMapMeta(carbonTable.getIndexedColumns(dataMapSchema),
+ Seq(ExpressionType.EQUALS).asJava)
override def toDistributable(segmentId: Segment): util.List[DataMapDistributable] = ???
- override def init(carbonTable: CarbonTable, dataMapSchema: DataMapSchema): Unit = {
- this.identifier = carbonTable.getAbsoluteTableIdentifier
- }
-
/**
* delete datamap data if any
*/
@@ -235,4 +244,9 @@ class TestDataMap() extends CoarseGrainDataMapFactory {
override def willBecomeStale(operation: TableOperation): Boolean = {
false
}
+
+ override def createRefresher(segment: Segment,
+ shardName: String): DataMapRefresher = {
+ ???
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertAndOtherCommandConcurrent.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertAndOtherCommandConcurrent.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertAndOtherCommandConcurrent.scala
index 12bec0a..9c5297d 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertAndOtherCommandConcurrent.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertAndOtherCommandConcurrent.scala
@@ -28,16 +28,14 @@ import org.apache.spark.sql.{DataFrame, SaveMode}
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.datamap.dev.DataMapWriter
+import org.apache.carbondata.core.datamap.dev.{DataMapRefresher, DataMapWriter}
import org.apache.carbondata.core.datamap.dev.cgdatamap.{CoarseGrainDataMap, CoarseGrainDataMapFactory}
-import org.apache.carbondata.core.datamap.{DataMapDistributable, DataMapMeta, DataMapStoreManager, Segment}
+import org.apache.carbondata.core.datamap.{DataMapDistributable, DataMapMeta, Segment}
import org.apache.carbondata.core.datastore.page.ColumnPage
import org.apache.carbondata.core.exception.ConcurrentOperationException
import org.apache.carbondata.core.features.TableOperation
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
-import org.apache.carbondata.core.metadata.schema.table.{DataMapSchema, RelationIdentifier}
-import org.apache.carbondata.core.readcommitter.ReadCommittedScope
-import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema, RelationIdentifier}
+import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}
import org.apache.carbondata.core.scan.filter.intf.ExpressionType
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.events.Event
@@ -54,8 +52,8 @@ class TestInsertAndOtherCommandConcurrent extends QueryTest with BeforeAndAfterA
sql(
s"""
| create datamap test on table orders
- | using '${classOf[WaitingDataMap].getName}'
- | as select count(a) from hiveMetaStoreTable_1")
+ | using '${classOf[WaitingDataMapFactory].getName}'
+ | dmproperties('index_columns'='o_name')
""".stripMargin)
}
@@ -110,7 +108,7 @@ class TestInsertAndOtherCommandConcurrent extends QueryTest with BeforeAndAfterA
)
while (!Global.overwriteRunning && count < 1000) {
Thread.sleep(10)
- // to avoid dead loop in case WaitingDataMap is not invoked
+ // to avoid dead loop in case WaitingDataMapFactory is not invoked
count += 1
}
future
@@ -211,8 +209,8 @@ class TestInsertAndOtherCommandConcurrent extends QueryTest with BeforeAndAfterA
sql(
s"""
| create datamap dm_t1 on table t1
- | using '${classOf[WaitingDataMap].getName}'
- | as select count(a) from hiveMetaStoreTable_1")
+ | using '${classOf[WaitingDataMapFactory].getName}'
+ | dmproperties('index_columns'='o_name')
""".stripMargin)
val future = runSqlAsync("insert into table t1 select * from orders_overwrite")
sql("alter table t1 compact 'MAJOR'")
@@ -284,9 +282,11 @@ object Global {
var overwriteRunning = false
}
-class WaitingDataMap() extends CoarseGrainDataMapFactory {
+class WaitingDataMapFactory(
+ carbonTable: CarbonTable,
+ dataMapSchema: DataMapSchema) extends CoarseGrainDataMapFactory(carbonTable, dataMapSchema) {
- private var identifier: AbsoluteTableIdentifier = _
+ private var identifier: AbsoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
override def fireEvent(event: Event): Unit = ???
@@ -298,9 +298,10 @@ class WaitingDataMap() extends CoarseGrainDataMapFactory {
override def getDataMaps(segment: Segment): util.List[CoarseGrainDataMap] = ???
- override def createWriter(segment: Segment, writeDirectoryPath: String): DataMapWriter = {
- new DataMapWriter(identifier, segment, writeDirectoryPath) {
- override def onPageAdded(blockletId: Int, pageId: Int, pages: Array[ColumnPage]): Unit = { }
+ override def createWriter(segment: Segment, shardName: String): DataMapWriter = {
+ new DataMapWriter(carbonTable.getTablePath, dataMapSchema.getDataMapName,
+ carbonTable.getIndexedColumns(dataMapSchema), segment, shardName) {
+ override def onPageAdded(blockletId: Int, pageId: Int, pageSize: Int, pages: Array[ColumnPage]): Unit = { }
override def onBlockletEnd(blockletId: Int): Unit = { }
@@ -308,7 +309,7 @@ class WaitingDataMap() extends CoarseGrainDataMapFactory {
override def onBlockletStart(blockletId: Int): Unit = { }
- override def onBlockStart(blockId: String, taskId: String): Unit = {
+ override def onBlockStart(blockId: String): Unit = {
// trigger the second SQL to execute
Global.overwriteRunning = true
@@ -322,15 +323,10 @@ class WaitingDataMap() extends CoarseGrainDataMapFactory {
}
}
- override def getMeta: DataMapMeta = new DataMapMeta(List("o_country").asJava, Seq(ExpressionType.EQUALS).asJava)
+ override def getMeta: DataMapMeta = new DataMapMeta(carbonTable.getIndexedColumns(dataMapSchema), Seq(ExpressionType.EQUALS).asJava)
override def toDistributable(segmentId: Segment): util.List[DataMapDistributable] = ???
- override def init(carbonTable: CarbonTable,
- dataMapSchema: DataMapSchema): Unit = {
- this.identifier = carbonTable.getAbsoluteTableIdentifier
- }
-
/**
* delete datamap data if any
*/
@@ -344,4 +340,9 @@ class WaitingDataMap() extends CoarseGrainDataMapFactory {
override def willBecomeStale(operation: TableOperation): Boolean = {
false
}
+
+ override def createRefresher(segment: Segment,
+ shardName: String): DataMapRefresher = {
+ ???
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark2/src/main/java/org/apache/carbondata/datamap/DataMapManager.java
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/java/org/apache/carbondata/datamap/DataMapManager.java b/integration/spark2/src/main/java/org/apache/carbondata/datamap/DataMapManager.java
index 886200b..3fe1f0d 100644
--- a/integration/spark2/src/main/java/org/apache/carbondata/datamap/DataMapManager.java
+++ b/integration/spark2/src/main/java/org/apache/carbondata/datamap/DataMapManager.java
@@ -17,8 +17,9 @@
package org.apache.carbondata.datamap;
+import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
import org.apache.carbondata.core.datamap.DataMapProvider;
-import org.apache.carbondata.core.datamap.IndexDataMapProvider;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
import static org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider.PREAGGREGATE;
@@ -42,15 +43,15 @@ public class DataMapManager {
/**
* Return a DataMapClassProvider instance for specified dataMapSchema.
*/
- public DataMapProvider getDataMapProvider(DataMapSchema dataMapSchema,
- SparkSession sparkSession) {
+ public DataMapProvider getDataMapProvider(CarbonTable mainTable, DataMapSchema dataMapSchema,
+ SparkSession sparkSession) throws MalformedDataMapCommandException {
DataMapProvider provider;
if (dataMapSchema.getProviderName().equalsIgnoreCase(PREAGGREGATE.toString())) {
- provider = new PreAggregateDataMapProvider(sparkSession);
+ provider = new PreAggregateDataMapProvider(mainTable, dataMapSchema, sparkSession);
} else if (dataMapSchema.getProviderName().equalsIgnoreCase(TIMESERIES.toString())) {
- provider = new TimeseriesDataMapProvider(sparkSession);
+ provider = new TimeseriesDataMapProvider(mainTable, dataMapSchema, sparkSession);
} else {
- provider = new IndexDataMapProvider();
+ provider = new IndexDataMapProvider(mainTable, dataMapSchema, sparkSession);
}
return provider;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark2/src/main/java/org/apache/carbondata/datamap/IndexDataMapProvider.java
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/java/org/apache/carbondata/datamap/IndexDataMapProvider.java b/integration/spark2/src/main/java/org/apache/carbondata/datamap/IndexDataMapProvider.java
new file mode 100644
index 0000000..a5124a0
--- /dev/null
+++ b/integration/spark2/src/main/java/org/apache/carbondata/datamap/IndexDataMapProvider.java
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.datamap;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.common.exceptions.MetadataProcessException;
+import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
+import org.apache.carbondata.core.datamap.DataMapCatalog;
+import org.apache.carbondata.core.datamap.DataMapProvider;
+import org.apache.carbondata.core.datamap.DataMapRegistry;
+import org.apache.carbondata.core.datamap.DataMapStoreManager;
+import org.apache.carbondata.core.datamap.dev.DataMap;
+import org.apache.carbondata.core.datamap.dev.DataMapFactory;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
+import org.apache.carbondata.core.metadata.schema.table.RelationIdentifier;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
+
+import org.apache.spark.sql.SparkSession;
+
+/**
+ * Index type DataMap, all index datamap should implement this interface.
+ */
+@InterfaceAudience.Internal
+public class IndexDataMapProvider extends DataMapProvider {
+
+ private SparkSession sparkSession;
+ private DataMapFactory<? extends DataMap> dataMapFactory;
+ private List<CarbonColumn> indexedColumns;
+
+ IndexDataMapProvider(CarbonTable table, DataMapSchema schema, SparkSession sparkSession)
+ throws MalformedDataMapCommandException {
+ super(table, schema);
+ this.sparkSession = sparkSession;
+ this.dataMapFactory = createDataMapFactory();
+ dataMapFactory.validate();
+ this.indexedColumns = table.getIndexedColumns(schema);
+ }
+
+ public List<CarbonColumn> getIndexedColumns() {
+ return indexedColumns;
+ }
+
+ @Override
+ public void initMeta(String ctasSqlStatement)
+ throws MalformedDataMapCommandException, IOException {
+ CarbonTable mainTable = getMainTable();
+ DataMapSchema dataMapSchema = getDataMapSchema();
+ if (mainTable == null) {
+ throw new MalformedDataMapCommandException(
+ "Parent table is required to create index datamap");
+ }
+ ArrayList<RelationIdentifier> relationIdentifiers = new ArrayList<>();
+ RelationIdentifier relationIdentifier =
+ new RelationIdentifier(mainTable.getDatabaseName(), mainTable.getTableName(),
+ mainTable.getTableInfo().getFactTable().getTableId());
+ relationIdentifiers.add(relationIdentifier);
+ dataMapSchema.setRelationIdentifier(relationIdentifier);
+ dataMapSchema.setParentTables(relationIdentifiers);
+ DataMapStoreManager.getInstance().registerDataMap(mainTable, dataMapSchema, dataMapFactory);
+ DataMapStoreManager.getInstance().saveDataMapSchema(dataMapSchema);
+ }
+
+ @Override
+ public void initData() {
+ // Nothing is needed to do by default
+ }
+
+ @Override
+ public void cleanMeta() throws IOException {
+ if (getMainTable() == null) {
+ throw new UnsupportedOperationException("Table need to be specified in index datamaps");
+ }
+ DataMapStoreManager.getInstance().dropDataMapSchema(getDataMapSchema().getDataMapName());
+ }
+
+ @Override
+ public void cleanData() {
+ CarbonTable mainTable = getMainTable();
+ if (mainTable == null) {
+ throw new UnsupportedOperationException("Table need to be specified in index datamaps");
+ }
+ DataMapStoreManager.getInstance().clearDataMap(
+ mainTable.getAbsoluteTableIdentifier(), getDataMapSchema().getDataMapName());
+ }
+
+ @Override
+ public void rebuild() {
+ IndexDataMapRefreshRDD.rebuildDataMap(sparkSession, getMainTable(), getDataMapSchema());
+ }
+
+ @Override
+ public void incrementalBuild(String[] segmentIds) {
+ throw new UnsupportedOperationException();
+ }
+
+ private DataMapFactory<? extends DataMap> createDataMapFactory()
+ throws MalformedDataMapCommandException {
+ CarbonTable mainTable = getMainTable();
+ DataMapSchema dataMapSchema = getDataMapSchema();
+ DataMapFactory<? extends DataMap> dataMapFactory;
+ try {
+ // try to create DataMapClassProvider instance by taking providerName as class name
+ dataMapFactory = (DataMapFactory<? extends DataMap>)
+ Class.forName(dataMapSchema.getProviderName()).getConstructors()[0]
+ .newInstance(mainTable, dataMapSchema);
+ } catch (ClassNotFoundException e) {
+ // try to create DataMapClassProvider instance by taking providerName as short name
+ dataMapFactory =
+ DataMapRegistry.getDataMapFactoryByShortName(mainTable, dataMapSchema);
+ } catch (Throwable e) {
+ throw new MetadataProcessException(
+ "failed to create DataMapClassProvider '" + dataMapSchema.getProviderName() + "'", e);
+ }
+ return dataMapFactory;
+ }
+
+ @Override
+ public DataMapCatalog createDataMapCatalog() {
+ // TODO create abstract class and move the default implementation there.
+ return null;
+ }
+
+ @Override
+ public DataMapFactory getDataMapFactory() {
+ return dataMapFactory;
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark2/src/main/java/org/apache/carbondata/datamap/PreAggregateDataMapProvider.java
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/java/org/apache/carbondata/datamap/PreAggregateDataMapProvider.java b/integration/spark2/src/main/java/org/apache/carbondata/datamap/PreAggregateDataMapProvider.java
index b612f47..746a361 100644
--- a/integration/spark2/src/main/java/org/apache/carbondata/datamap/PreAggregateDataMapProvider.java
+++ b/integration/spark2/src/main/java/org/apache/carbondata/datamap/PreAggregateDataMapProvider.java
@@ -21,6 +21,7 @@ import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
import org.apache.carbondata.core.datamap.DataMapCatalog;
import org.apache.carbondata.core.datamap.DataMapProvider;
+import org.apache.carbondata.core.datamap.dev.DataMapFactory;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
@@ -30,21 +31,23 @@ import org.apache.spark.sql.execution.command.table.CarbonDropTableCommand;
import scala.Some;
@InterfaceAudience.Internal
-public class PreAggregateDataMapProvider implements DataMapProvider {
+public class PreAggregateDataMapProvider extends DataMapProvider {
protected PreAggregateTableHelper helper;
protected CarbonDropTableCommand dropTableCommand;
protected SparkSession sparkSession;
- public PreAggregateDataMapProvider(SparkSession sparkSession) {
+ PreAggregateDataMapProvider(CarbonTable table, DataMapSchema schema,
+ SparkSession sparkSession) {
+ super(table, schema);
this.sparkSession = sparkSession;
}
@Override
- public void initMeta(CarbonTable mainTable, DataMapSchema dataMapSchema, String ctasSqlStatement)
- throws MalformedDataMapCommandException {
+ public void initMeta(String ctasSqlStatement) throws MalformedDataMapCommandException {
+ DataMapSchema dataMapSchema = getDataMapSchema();
validateDmProperty(dataMapSchema);
helper = new PreAggregateTableHelper(
- mainTable, dataMapSchema.getDataMapName(), dataMapSchema.getProviderName(),
+ getMainTable(), dataMapSchema.getDataMapName(), dataMapSchema.getProviderName(),
dataMapSchema.getProperties(), ctasSqlStatement, null, false);
helper.initMeta(sparkSession);
}
@@ -62,12 +65,13 @@ public class PreAggregateDataMapProvider implements DataMapProvider {
}
@Override
- public void initData(CarbonTable mainTable) {
+ public void initData() {
// Nothing is needed to do by default
}
@Override
- public void freeMeta(CarbonTable mainTable, DataMapSchema dataMapSchema) {
+ public void cleanMeta() {
+ DataMapSchema dataMapSchema = getDataMapSchema();
dropTableCommand = new CarbonDropTableCommand(
true,
new Some<>(dataMapSchema.getRelationIdentifier().getDatabaseName()),
@@ -77,21 +81,20 @@ public class PreAggregateDataMapProvider implements DataMapProvider {
}
@Override
- public void freeData(CarbonTable mainTable, DataMapSchema dataMapSchema) {
+ public void cleanData() {
if (dropTableCommand != null) {
dropTableCommand.processData(sparkSession);
}
}
@Override
- public void rebuild(CarbonTable mainTable, DataMapSchema dataMapSchema) {
+ public void rebuild() {
if (helper != null) {
helper.initData(sparkSession);
}
}
- @Override public void incrementalBuild(CarbonTable mainTable, DataMapSchema dataMapSchema,
- String[] segmentIds) {
+ @Override public void incrementalBuild(String[] segmentIds) {
throw new UnsupportedOperationException();
}
@@ -99,4 +102,9 @@ public class PreAggregateDataMapProvider implements DataMapProvider {
// TODO manage pre-agg also with catalog.
return null;
}
+
+ @Override
+ public DataMapFactory getDataMapFactory() {
+ throw new UnsupportedOperationException();
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark2/src/main/java/org/apache/carbondata/datamap/TimeseriesDataMapProvider.java
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/java/org/apache/carbondata/datamap/TimeseriesDataMapProvider.java b/integration/spark2/src/main/java/org/apache/carbondata/datamap/TimeseriesDataMapProvider.java
index c2acca4..4b93f59 100644
--- a/integration/spark2/src/main/java/org/apache/carbondata/datamap/TimeseriesDataMapProvider.java
+++ b/integration/spark2/src/main/java/org/apache/carbondata/datamap/TimeseriesDataMapProvider.java
@@ -32,13 +32,15 @@ import scala.Tuple2;
@InterfaceAudience.Internal
public class TimeseriesDataMapProvider extends PreAggregateDataMapProvider {
- public TimeseriesDataMapProvider(SparkSession sparkSession) {
- super(sparkSession);
+ TimeseriesDataMapProvider(CarbonTable mainTable, DataMapSchema dataMapSchema,
+ SparkSession sparkSession) {
+ super(mainTable, dataMapSchema, sparkSession);
}
@Override
- public void initMeta(CarbonTable mainTable, DataMapSchema dataMapSchema,
- String ctasSqlStatement) {
+ public void initMeta(String ctasSqlStatement) {
+ DataMapSchema dataMapSchema = getDataMapSchema();
+ CarbonTable mainTable = getMainTable();
Map<String, String> dmProperties = dataMapSchema.getProperties();
String dmProviderName = dataMapSchema.getProviderName();
TimeSeriesUtil.validateTimeSeriesGranularity(dmProperties, dmProviderName);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRefreshRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRefreshRDD.scala b/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRefreshRDD.scala
new file mode 100644
index 0000000..c341c36
--- /dev/null
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRefreshRDD.scala
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.datamap
+
+import java.io.{File, IOException}
+import java.text.SimpleDateFormat
+import java.util
+
+import scala.collection.JavaConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hadoop.mapreduce.{Job, TaskAttemptID, TaskType}
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
+import org.apache.spark.{CarbonInputMetrics, Partition, TaskContext}
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.sql.SparkSession
+
+import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.datamap.{DataMapStoreManager, Segment}
+import org.apache.carbondata.core.datamap.dev.DataMapRefresher
+import org.apache.carbondata.core.datastore.impl.FileFactory
+import org.apache.carbondata.core.metadata.datatype.{DataType, DataTypes}
+import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema, TableInfo}
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn
+import org.apache.carbondata.core.statusmanager.SegmentStatusManager
+import org.apache.carbondata.core.util.TaskMetricsMap
+import org.apache.carbondata.core.util.path.CarbonTablePath
+import org.apache.carbondata.hadoop.{CarbonInputSplit, CarbonMultiBlockSplit, CarbonProjection, CarbonRecordReader}
+import org.apache.carbondata.hadoop.api.{CarbonInputFormat, CarbonTableInputFormat}
+import org.apache.carbondata.hadoop.readsupport.CarbonReadSupport
+import org.apache.carbondata.spark.{RefreshResult, RefreshResultImpl}
+import org.apache.carbondata.spark.rdd.{CarbonRDDWithTableInfo, CarbonSparkPartition}
+import org.apache.carbondata.spark.util.SparkDataTypeConverterImpl
+
+/**
+ * Helper object to rebuild the index DataMap
+ */
+object IndexDataMapRefreshRDD {
+
+ /**
+ * Rebuild the datamap for all existing data in the table
+ */
+ def rebuildDataMap(
+ sparkSession: SparkSession,
+ carbonTable: CarbonTable,
+ schema: DataMapSchema
+ ): Unit = {
+ val tableIdentifier = carbonTable.getAbsoluteTableIdentifier
+ val segmentStatusManager = new SegmentStatusManager(tableIdentifier)
+ val validAndInvalidSegments = segmentStatusManager.getValidAndInvalidSegments()
+ val validSegments = validAndInvalidSegments.getValidSegments
+ val indexedCarbonColumns = carbonTable.getIndexedColumns(schema)
+
+ // loop all segments to rebuild DataMap
+ validSegments.asScala.foreach { segment =>
+ // if lucene datamap folder is exists, not require to build lucene datamap again
+ refreshOneSegment(sparkSession, carbonTable, schema.getDataMapName,
+ indexedCarbonColumns, segment.getSegmentNo);
+ }
+ }
+
+ private def refreshOneSegment(
+ sparkSession: SparkSession,
+ carbonTable: CarbonTable,
+ dataMapName: String,
+ indexColumns: java.util.List[CarbonColumn],
+ segmentId: String): Unit = {
+
+ val dataMapStorePath =
+ CarbonTablePath.getSegmentPath(carbonTable.getTablePath, segmentId) +
+ File.separator +
+ dataMapName
+
+ if (!FileFactory.isFileExist(dataMapStorePath)) {
+ if (FileFactory.mkdirs(dataMapStorePath, FileFactory.getFileType(dataMapStorePath))) {
+ try {
+ val status = new IndexDataMapRefreshRDD[String, Boolean](
+ sparkSession,
+ new RefreshResultImpl(),
+ carbonTable.getTableInfo,
+ dataMapName,
+ indexColumns.asScala.toArray,
+ segmentId
+ ).collect()
+
+ status.find(_._2 == false).foreach { task =>
+ throw new Exception(
+ s"Task Failed to refresh datamap $dataMapName on segment_$segmentId")
+ }
+ } catch {
+ case ex: Throwable =>
+ // process failure
+ FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(dataMapStorePath))
+ throw new Exception(
+ s"Failed to refresh datamap $dataMapName on segment_$segmentId", ex)
+ }
+ } else {
+ throw new IOException(s"Failed to create directory $dataMapStorePath")
+ }
+ }
+ }
+
+}
+
+class OriginalReadSupport(dataTypes: Array[DataType]) extends CarbonReadSupport[Array[Object]] {
+ override def initialize(carbonColumns: Array[CarbonColumn],
+ carbonTable: CarbonTable): Unit = {
+ }
+
+ override def readRow(data: Array[Object]): Array[Object] = {
+ dataTypes.zipWithIndex.foreach { case (dataType, i) =>
+ if (dataType == DataTypes.STRING) {
+ data(i) = data(i).toString
+ }
+ }
+ data
+ }
+
+ override def close(): Unit = {
+ }
+}
+
+class IndexDataMapRefreshRDD[K, V](
+ session: SparkSession,
+ result: RefreshResult[K, V],
+ @transient tableInfo: TableInfo,
+ dataMapName: String,
+ indexColumns: Array[CarbonColumn],
+ segmentId: String
+) extends CarbonRDDWithTableInfo[(K, V)](
+ session.sparkContext, Nil, tableInfo.serialize()) {
+
+ private val dataMapSchema = DataMapStoreManager.getInstance().getDataMapSchema(dataMapName)
+ private val queryId = sparkContext.getConf.get("queryId", System.nanoTime() + "")
+ private val jobTrackerId: String = {
+ val formatter = new SimpleDateFormat("yyyyMMddHHmm")
+ formatter.format(new util.Date())
+ }
+
+ override def internalCompute(split: Partition, context: TaskContext): Iterator[(K, V)] = {
+ val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
+ val dataMapFactory =
+ DataMapManager.get().getDataMapProvider(
+ CarbonTable.buildFromTableInfo(getTableInfo), dataMapSchema, session).getDataMapFactory
+ var status = false
+ val inputMetrics = new CarbonInputMetrics
+ TaskMetricsMap.getInstance().registerThreadCallback()
+ val inputSplit = split.asInstanceOf[CarbonSparkPartition].split.value
+ inputMetrics.initBytesReadCallback(context, inputSplit)
+
+ val attemptId = new TaskAttemptID(jobTrackerId, id, TaskType.MAP, split.index, 0)
+ val attemptContext = new TaskAttemptContextImpl(new Configuration(), attemptId)
+ val format = createInputFormat(attemptContext)
+
+ val model = format.createQueryModel(inputSplit, attemptContext)
+ // one query id per table
+ model.setQueryId(queryId)
+ model.setVectorReader(false)
+ model.setForcedDetailRawQuery(false)
+ model.setRequiredRowId(true)
+
+ var reader: CarbonRecordReader[Array[Object]] = null
+ var refresher: DataMapRefresher = null
+ try {
+ reader = new CarbonRecordReader(
+ model, new OriginalReadSupport(indexColumns.map(_.getDataType)), inputMetrics)
+ reader.initialize(inputSplit, attemptContext)
+
+ // we use task name as shard name to create the folder for this datamap
+ val shardName = CarbonTablePath.getShardName(inputSplit.getAllSplits.get(0).getBlockPath)
+ refresher = dataMapFactory.createRefresher(new Segment(segmentId), shardName)
+ refresher.initialize()
+
+ var blockletId = 0
+ var firstRow = true
+ while (reader.nextKeyValue()) {
+ val rowWithPosition = reader.getCurrentValue
+ val size = rowWithPosition.length
+ val pageId = rowWithPosition(size - 2).asInstanceOf[Int]
+ val rowId = rowWithPosition(size - 1).asInstanceOf[Int]
+
+ if (!firstRow && pageId == 0 && rowId == 0) {
+ // new blocklet started, increase blockletId
+ blockletId = blockletId + 1
+ } else {
+ firstRow = false
+ }
+
+ refresher.addRow(blockletId, pageId, rowId, rowWithPosition)
+ }
+
+ refresher.finish()
+
+ status = true
+ } finally {
+ if (reader != null) {
+ try {
+ reader.close()
+ } catch {
+ case ex: Throwable =>
+ LOGGER.error(ex, "Failed to close reader")
+ }
+ }
+
+ if (refresher != null) {
+ try {
+ refresher.close()
+ } catch {
+ case ex: Throwable =>
+ LOGGER.error(ex, "Failed to close index writer")
+ }
+ }
+ }
+
+ new Iterator[(K, V)] {
+
+ var finished = false
+
+ override def hasNext: Boolean = {
+ !finished
+ }
+
+ override def next(): (K, V) = {
+ finished = true
+ result.getKey(split.index.toString, status)
+ }
+ }
+ }
+
+
+ private def createInputFormat(
+ attemptContext: TaskAttemptContextImpl) = {
+ val format = new CarbonTableInputFormat[Object]
+ val tableInfo1 = getTableInfo
+ val conf = attemptContext.getConfiguration
+ CarbonInputFormat.setTableInfo(conf, tableInfo1)
+ CarbonInputFormat.setDatabaseName(conf, tableInfo1.getDatabaseName)
+ CarbonInputFormat.setTableName(conf, tableInfo1.getFactTable.getTableName)
+ CarbonInputFormat.setDataTypeConverter(conf, classOf[SparkDataTypeConverterImpl])
+
+ val identifier = tableInfo1.getOrCreateAbsoluteTableIdentifier()
+ CarbonInputFormat.setTablePath(
+ conf,
+ identifier.appendWithLocalPrefix(identifier.getTablePath))
+
+ CarbonInputFormat.setSegmentsToAccess(
+ conf,
+ Segment.toSegmentList(Array(segmentId), null))
+
+ CarbonInputFormat.setColumnProjection(
+ conf,
+ new CarbonProjection(indexColumns.map(_.getColName)))
+ format
+ }
+
+ override protected def getPartitions = {
+ if (!dataMapSchema.isIndexDataMap) {
+ throw new UnsupportedOperationException
+ }
+ val conf = new Configuration()
+ val jobConf = new JobConf(conf)
+ SparkHadoopUtil.get.addCredentials(jobConf)
+ val job = Job.getInstance(jobConf)
+ job.getConfiguration.set("query.id", queryId)
+
+ val format = new CarbonTableInputFormat[Object]
+
+ CarbonInputFormat.setSegmentsToAccess(
+ job.getConfiguration,
+ Segment.toSegmentList(Array(segmentId), null))
+
+ CarbonInputFormat.setTableInfo(
+ job.getConfiguration,
+ tableInfo)
+ CarbonInputFormat.setTablePath(
+ job.getConfiguration,
+ tableInfo.getOrCreateAbsoluteTableIdentifier().getTablePath)
+ CarbonInputFormat.setDatabaseName(
+ job.getConfiguration,
+ tableInfo.getDatabaseName)
+ CarbonInputFormat.setTableName(
+ job.getConfiguration,
+ tableInfo.getFactTable.getTableName)
+
+ format
+ .getSplits(job)
+ .asScala
+ .map(_.asInstanceOf[CarbonInputSplit])
+ .groupBy(_.taskId)
+ .map { group =>
+ new CarbonMultiBlockSplit(
+ group._2.asJava,
+ group._2.flatMap(_.getLocations).toArray)
+ }
+ .zipWithIndex
+ .map { split =>
+ new CarbonSparkPartition(id, split._2, split._1)
+ }
+ .toArray
+ }
+}
[3/4] carbondata git commit: [CARBONDATA-2415] Support Refresh
DataMap command for all Index datamap
Posted by ra...@apache.org.
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDataMapWriter.java
----------------------------------------------------------------------
diff --git a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDataMapWriter.java b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDataMapWriter.java
index 76ee084..4e07182 100644
--- a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDataMapWriter.java
+++ b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomDataMapWriter.java
@@ -17,85 +17,74 @@
package org.apache.carbondata.datamap.bloom;
import java.io.DataOutputStream;
-import java.io.File;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
-import org.apache.carbondata.core.datamap.DataMapMeta;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.datastore.page.ColumnPage;
-import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.util.CarbonUtil;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
/**
- * BloomDataMap is constructed in blocklet level. For each indexed column, a bloom filter is
- * constructed to indicate whether a value belongs to this blocklet. Bloom filter of blocklet that
- * belongs to same block will be written to one index file suffixed with .bloomindex. So the number
+ * BloomDataMap is constructed in CG level (blocklet level).
+ * For each indexed column, a bloom filter is constructed to indicate whether a value
+ * belongs to this blocklet. Bloom filter of blocklet that belongs to same block will
+ * be written to one index file suffixed with .bloomindex. So the number
* of bloom index file will be equal to that of the blocks.
*/
@InterfaceAudience.Internal
public class BloomDataMapWriter extends DataMapWriter {
- private String dataMapName;
- private List<String> indexedColumns;
+ private static final LogService LOG = LogServiceFactory.getLogService(
+ BloomDataMapWriter.class.getCanonicalName());
private int bloomFilterSize;
- // map column name to ordinal in pages
- private Map<String, Integer> col2Ordianl;
- private Map<String, DataType> col2DataType;
- private String indexShardName;
- private int currentBlockletId;
+ protected int currentBlockletId;
private List<String> currentDMFiles;
private List<DataOutputStream> currentDataOutStreams;
private List<ObjectOutputStream> currentObjectOutStreams;
- private List<BloomFilter<byte[]>> indexBloomFilters;
-
- @InterfaceAudience.Internal
- public BloomDataMapWriter(AbsoluteTableIdentifier identifier, DataMapMeta dataMapMeta,
- int bloomFilterSize, Segment segment, String writeDirectoryPath) {
- super(identifier, segment, writeDirectoryPath);
- dataMapName = dataMapMeta.getDataMapName();
- indexedColumns = dataMapMeta.getIndexedColumns();
- this.bloomFilterSize = bloomFilterSize;
- col2Ordianl = new HashMap<String, Integer>(indexedColumns.size());
- col2DataType = new HashMap<String, DataType>(indexedColumns.size());
+ protected List<BloomFilter<byte[]>> indexBloomFilters;
- currentDMFiles = new ArrayList<String>(indexedColumns.size());
- currentDataOutStreams = new ArrayList<DataOutputStream>(indexedColumns.size());
- currentObjectOutStreams = new ArrayList<ObjectOutputStream>(indexedColumns.size());
+ BloomDataMapWriter(String tablePath, String dataMapName, List<CarbonColumn> indexColumns,
+ Segment segment, String shardName, int bloomFilterSize) throws IOException {
+ super(tablePath, dataMapName, indexColumns, segment, shardName);
+ this.bloomFilterSize = bloomFilterSize;
- indexBloomFilters = new ArrayList<BloomFilter<byte[]>>(indexedColumns.size());
+ currentDMFiles = new ArrayList<String>(indexColumns.size());
+ currentDataOutStreams = new ArrayList<DataOutputStream>(indexColumns.size());
+ currentObjectOutStreams = new ArrayList<ObjectOutputStream>(indexColumns.size());
+ indexBloomFilters = new ArrayList<BloomFilter<byte[]>>(indexColumns.size());
+ initDataMapFile();
+ resetBloomFilters();
}
@Override
- public void onBlockStart(String blockId, String indexShardName) throws IOException {
- if (this.indexShardName == null) {
- this.indexShardName = indexShardName;
- initDataMapFile();
- }
+ public void onBlockStart(String blockId) throws IOException {
}
@Override
public void onBlockEnd(String blockId) throws IOException {
-
}
@Override
public void onBlockletStart(int blockletId) {
- this.currentBlockletId = blockletId;
+ }
+
+ protected void resetBloomFilters() {
indexBloomFilters.clear();
- for (int i = 0; i < indexedColumns.size(); i++) {
+ List<CarbonColumn> indexColumns = getIndexColumns();
+ for (int i = 0; i < indexColumns.size(); i++) {
indexBloomFilters.add(BloomFilter.create(Funnels.byteArrayFunnel(),
bloomFilterSize, 0.00001d));
}
@@ -103,60 +92,51 @@ public class BloomDataMapWriter extends DataMapWriter {
@Override
public void onBlockletEnd(int blockletId) {
- try {
- writeBloomDataMapFile();
- } catch (Exception e) {
- for (ObjectOutputStream objectOutputStream : currentObjectOutStreams) {
- CarbonUtil.closeStreams(objectOutputStream);
- }
- for (DataOutputStream dataOutputStream : currentDataOutStreams) {
- CarbonUtil.closeStreams(dataOutputStream);
- }
- throw new RuntimeException(e);
- }
+ writeBloomDataMapFile();
+ currentBlockletId++;
}
- // notice that the input pages only contains the indexed columns
@Override
- public void onPageAdded(int blockletId, int pageId, ColumnPage[] pages)
- throws IOException {
- col2Ordianl.clear();
- col2DataType.clear();
- for (int colId = 0; colId < pages.length; colId++) {
- String columnName = pages[colId].getColumnSpec().getFieldName().toLowerCase();
- col2Ordianl.put(columnName, colId);
- DataType columnType = pages[colId].getColumnSpec().getSchemaDataType();
- col2DataType.put(columnName, columnType);
- }
-
- // for each row
- for (int rowId = 0; rowId < pages[0].getPageSize(); rowId++) {
- // for each indexed column
- for (int indexColId = 0; indexColId < indexedColumns.size(); indexColId++) {
- String indexedCol = indexedColumns.get(indexColId);
+ public void onPageAdded(int blockletId, int pageId, int pageSize, ColumnPage[] pages) {
+ List<CarbonColumn> indexColumns = getIndexColumns();
+ for (int rowId = 0; rowId < pageSize; rowId++) {
+ // for each indexed column, add the data to bloom filter
+ for (int i = 0; i < indexColumns.size(); i++) {
+ Object data = pages[i].getData(rowId);
+ DataType dataType = indexColumns.get(i).getDataType();
byte[] indexValue;
- if (DataTypes.STRING == col2DataType.get(indexedCol)
- || DataTypes.BYTE_ARRAY == col2DataType.get(indexedCol)) {
- byte[] originValue = (byte[]) pages[col2Ordianl.get(indexedCol)].getData(rowId);
+ if (DataTypes.STRING == dataType) {
+ indexValue = getStringData(data);
+ } else if (DataTypes.BYTE_ARRAY == dataType) {
+ byte[] originValue = (byte[]) data;
+ // String and byte array is LV encoded, L is short type
indexValue = new byte[originValue.length - 2];
System.arraycopy(originValue, 2, indexValue, 0, originValue.length - 2);
} else {
- Object originValue = pages[col2Ordianl.get(indexedCol)].getData(rowId);
- indexValue = CarbonUtil.getValueAsBytes(col2DataType.get(indexedCol), originValue);
+ indexValue = CarbonUtil.getValueAsBytes(dataType, data);
}
-
- indexBloomFilters.get(indexColId).put(indexValue);
+ indexBloomFilters.get(i).put(indexValue);
}
}
}
+ protected byte[] getStringData(Object data) {
+ byte[] lvData = (byte[]) data;
+ byte[] indexValue = new byte[lvData.length - 2];
+ System.arraycopy(lvData, 2, indexValue, 0, lvData.length - 2);
+ return indexValue;
+ }
+
private void initDataMapFile() throws IOException {
- String dataMapDir = genDataMapStorePath(this.writeDirectoryPath, this.dataMapName);
- dataMapDir = dataMapDir + CarbonCommonConstants.FILE_SEPARATOR + this.indexShardName;
- FileFactory.mkdirs(dataMapDir, FileFactory.getFileType(dataMapDir));
- for (int indexColId = 0; indexColId < indexedColumns.size(); indexColId++) {
- String dmFile = dataMapDir + CarbonCommonConstants.FILE_SEPARATOR +
- indexedColumns.get(indexColId) + BloomCoarseGrainDataMap.BLOOM_INDEX_SUFFIX;
+ if (!FileFactory.isFileExist(dataMapPath)) {
+ if (!FileFactory.mkdirs(dataMapPath, FileFactory.getFileType(dataMapPath))) {
+ throw new IOException("Failed to create directory " + dataMapPath);
+ }
+ }
+ List<CarbonColumn> indexColumns = getIndexColumns();
+ for (int indexColId = 0; indexColId < indexColumns.size(); indexColId++) {
+ String dmFile = dataMapPath + CarbonCommonConstants.FILE_SEPARATOR +
+ indexColumns.get(indexColId).getColName() + BloomCoarseGrainDataMap.BLOOM_INDEX_SUFFIX;
DataOutputStream dataOutStream = null;
ObjectOutputStream objectOutStream = null;
try {
@@ -175,44 +155,45 @@ public class BloomDataMapWriter extends DataMapWriter {
}
}
- private void writeBloomDataMapFile() throws IOException {
- for (int indexColId = 0; indexColId < indexedColumns.size(); indexColId++) {
- BloomDMModel model = new BloomDMModel(this.currentBlockletId,
- indexBloomFilters.get(indexColId));
- // only in higher version of guava-bloom-filter, it provides readFrom/writeTo interface.
- // In lower version, we use default java serializer to write bloomfilter.
- this.currentObjectOutStreams.get(indexColId).writeObject(model);
- this.currentObjectOutStreams.get(indexColId).flush();
- this.currentDataOutStreams.get(indexColId).flush();
+ protected void writeBloomDataMapFile() {
+ List<CarbonColumn> indexColumns = getIndexColumns();
+ try {
+ for (int indexColId = 0; indexColId < indexColumns.size(); indexColId++) {
+ BloomDMModel model =
+ new BloomDMModel(this.currentBlockletId, indexBloomFilters.get(indexColId));
+ // only in higher version of guava-bloom-filter, it provides readFrom/writeTo interface.
+ // In lower version, we use default java serializer to write bloomfilter.
+ this.currentObjectOutStreams.get(indexColId).writeObject(model);
+ this.currentObjectOutStreams.get(indexColId).flush();
+ this.currentDataOutStreams.get(indexColId).flush();
+ }
+ } catch (Exception e) {
+ for (ObjectOutputStream objectOutputStream : currentObjectOutStreams) {
+ CarbonUtil.closeStreams(objectOutputStream);
+ }
+ for (DataOutputStream dataOutputStream : currentDataOutStreams) {
+ CarbonUtil.closeStreams(dataOutputStream);
+ }
+ throw new RuntimeException(e);
+ } finally {
+ resetBloomFilters();
}
}
@Override
public void finish() throws IOException {
- for (int indexColId = 0; indexColId < indexedColumns.size(); indexColId++) {
- CarbonUtil.closeStreams(this.currentDataOutStreams.get(indexColId),
- this.currentObjectOutStreams.get(indexColId));
- commitFile(this.currentDMFiles.get(indexColId));
+ if (indexBloomFilters.size() > 0) {
+ writeBloomDataMapFile();
}
+ releaseResouce();
}
- @Override
- protected void commitFile(String dataMapFile) throws IOException {
- super.commitFile(dataMapFile);
+ protected void releaseResouce() {
+ List<CarbonColumn> indexColumns = getIndexColumns();
+ for (int indexColId = 0; indexColId < indexColumns.size(); indexColId++) {
+ CarbonUtil.closeStreams(
+ currentDataOutStreams.get(indexColId), currentObjectOutStreams.get(indexColId));
+ }
}
- /**
- * create and return path that will store the datamap
- *
- * @param dataPath patch to store the carbondata factdata
- * @param dataMapName datamap name
- * @return path to store the datamap
- * @throws IOException
- */
- public static String genDataMapStorePath(String dataPath, String dataMapName)
- throws IOException {
- String dmDir = dataPath + File.separator + dataMapName;
- FileFactory.mkdirs(dmDir, FileFactory.getFileType(dmDir));
- return dmDir;
- }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxDataWriter.java
----------------------------------------------------------------------
diff --git a/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxDataWriter.java b/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxDataWriter.java
index 21a0b8e..0993218 100644
--- a/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxDataWriter.java
+++ b/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxDataWriter.java
@@ -33,15 +33,12 @@ import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.datastore.page.ColumnPage;
-import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
-import org.apache.carbondata.core.metadata.CarbonMetadata;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
import org.apache.carbondata.core.metadata.schema.table.TableInfo;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
-import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
-import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure;
import org.apache.carbondata.core.util.ByteUtil;
import org.apache.carbondata.core.util.CarbonUtil;
@@ -58,7 +55,6 @@ public class MinMaxDataWriter extends DataMapWriter {
private Map<Integer, BlockletMinMax> blockMinMaxMap;
- private String dataMapName;
private int columnCnt;
private DataType[] dataTypeArray;
private String indexShardName;
@@ -71,30 +67,25 @@ public class MinMaxDataWriter extends DataMapWriter {
*/
private Map<Integer, Integer> origin2MinMaxOrdinal = new HashMap<>();
- public MinMaxDataWriter(AbsoluteTableIdentifier identifier, String dataMapName, Segment segment,
- String dataWritePath) {
- super(identifier, segment, dataWritePath);
- this.dataMapName = dataMapName;
- CarbonTable carbonTable = CarbonMetadata.getInstance().getCarbonTable(
- identifier.getDatabaseName(), identifier.getTableName());
- List<CarbonColumn> cols = carbonTable.getCreateOrderColumn(identifier.getTableName());
- this.columnCnt = cols.size();
- List<CarbonDimension> dimensions = carbonTable.getDimensionByTableName(identifier.getTableName());
- for (int i = 0; i < dimensions.size(); i++) {
- this.origin2MinMaxOrdinal.put(dimensions.get(i).getSchemaOrdinal(),
- dimensions.get(i).getOrdinal());
+ public MinMaxDataWriter(CarbonTable carbonTable, DataMapSchema dataMapSchema, Segment segment,
+ String shardName, List<CarbonColumn> indexColumns) {
+ super(carbonTable.getTablePath(), dataMapSchema.getDataMapName(), indexColumns, segment,
+ shardName);
+ this.columnCnt = indexColumns.size();
+ for (CarbonColumn col : indexColumns) {
+ this.origin2MinMaxOrdinal.put(col.getSchemaOrdinal(), col.getOrdinal());
}
- List<CarbonMeasure> measures = carbonTable.getMeasureByTableName(identifier.getTableName());
- for (int i = 0; i < measures.size(); i++) {
- this.origin2MinMaxOrdinal.put(measures.get(i).getSchemaOrdinal(),
- dimensions.size() + measures.get(i).getOrdinal());
+ if (this.dataTypeArray == null) {
+ this.dataTypeArray = new DataType[this.columnCnt];
+ for (int i = 0; i < this.columnCnt; i++) {
+ this.dataTypeArray[i] = indexColumns.get(i).getDataType();
+ }
}
}
- @Override public void onBlockStart(String blockId, String indexShardName) {
+ @Override public void onBlockStart(String blockId) {
if (blockMinMaxMap == null) {
blockMinMaxMap = new HashMap<>();
- this.indexShardName = indexShardName;
}
}
@@ -111,46 +102,15 @@ public class MinMaxDataWriter extends DataMapWriter {
}
@Override
- public void onPageAdded(int blockletId, int pageId, ColumnPage[] pages) {
- // Calculate Min and Max value within this page.
-
- // As part of example we are extracting Min Max values Manually. The same can be done from
- // retrieving the page statistics. For e.g.
-
- // if (pageLevelMin == null && pageLevelMax == null) {
- // pageLevelMin[1] = CarbonUtil.getValueAsBytes(pages[0].getStatistics().getDataType(),
- // pages[0].getStatistics().getMin());
- // pageLevelMax[1] = CarbonUtil.getValueAsBytes(pages[0].getStatistics().getDataType(),
- // pages[0].getStatistics().getMax());
- // } else {
- // if (ByteUtil.UnsafeComparer.INSTANCE.compareTo(pageLevelMin[1], CarbonUtil
- // .getValueAsBytes(pages[0].getStatistics().getDataType(),
- // pages[0].getStatistics().getMin())) > 0) {
- // pageLevelMin[1] = CarbonUtil.getValueAsBytes(pages[0].getStatistics().getDataType(),
- // pages[0].getStatistics().getMin());
- // }
- // if (ByteUtil.UnsafeComparer.INSTANCE.compareTo(pageLevelMax[1], CarbonUtil
- // .getValueAsBytes(pages[0].getStatistics().getDataType(),
- // pages[0].getStatistics().getMax())) < 0) {
- // pageLevelMax[1] = CarbonUtil.getValueAsBytes(pages[0].getStatistics().getDataType(),
- // pages[0].getStatistics().getMax());
- // }
-
- if (this.dataTypeArray == null) {
- this.dataTypeArray = new DataType[this.columnCnt];
- for (int i = 0; i < this.columnCnt; i++) {
- this.dataTypeArray[i] = pages[i].getDataType();
- }
- }
-
+ public void onPageAdded(int blockletId, int pageId, int pageSize, ColumnPage[] pages) {
// as an example, we don't use page-level min-max generated by native carbondata here, we get
// the min-max by comparing each row
- for (int rowId = 0; rowId < pages[0].getPageSize(); rowId++) {
+ for (int rowId = 0; rowId < pageSize; rowId++) {
for (int colIdx = 0; colIdx < columnCnt; colIdx++) {
Object originValue = pages[colIdx].getData(rowId);
+ DataType dataType = dataTypeArray[colIdx];
// for string & bytes_array, data is prefixed with length, need to remove it
- if (DataTypes.STRING == pages[colIdx].getDataType()
- || DataTypes.BYTE_ARRAY == pages[colIdx].getDataType()) {
+ if (DataTypes.STRING == dataType || DataTypes.BYTE_ARRAY == dataType) {
byte[] valueMin0 = (byte[]) pageLevelMin[colIdx];
byte[] valueMax0 = (byte[]) pageLevelMax[colIdx];
byte[] value1 = (byte[]) originValue;
@@ -164,10 +124,10 @@ public class MinMaxDataWriter extends DataMapWriter {
pageLevelMax[colIdx] = new byte[value1.length - 2];
System.arraycopy(value1, 2, (byte[]) pageLevelMax[colIdx], 0, value1.length - 2);
}
- } else if (DataTypes.INT == pages[colIdx].getDataType()) {
- updateMinMax(colIdx, originValue, pages[colIdx].getDataType());
+ } else if (DataTypes.INT == dataType) {
+ updateMinMax(colIdx, originValue, dataType);
} else {
- throw new RuntimeException("Not implement yet");
+ throw new UnsupportedOperationException("Not implement yet");
}
}
}
@@ -276,8 +236,7 @@ public class MinMaxDataWriter extends DataMapWriter {
*/
public void writeMinMaxIndexFile(List<MinMaxIndexBlockDetails> minMaxIndexBlockDetails,
String blockId) throws IOException {
- String dataMapDir = genDataMapStorePath(this.writeDirectoryPath, this.dataMapName);
- String filePath = dataMapDir + File.separator + blockId + ".minmaxindex";
+ String filePath = dataMapPath + File.separator + blockId + ".minmaxindex";
BufferedWriter brWriter = null;
DataOutputStream dataOutStream = null;
try {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxIndexDataMapFactory.java
----------------------------------------------------------------------
diff --git a/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxIndexDataMapFactory.java b/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxIndexDataMapFactory.java
index c110887..4197b79 100644
--- a/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxIndexDataMapFactory.java
+++ b/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxIndexDataMapFactory.java
@@ -28,6 +28,7 @@ import org.apache.carbondata.core.datamap.DataMapDistributable;
import org.apache.carbondata.core.datamap.DataMapMeta;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.dev.DataMapModel;
+import org.apache.carbondata.core.datamap.dev.DataMapRefresher;
import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap;
import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMapFactory;
@@ -51,25 +52,25 @@ import org.apache.commons.lang3.StringUtils;
public class MinMaxIndexDataMapFactory extends CoarseGrainDataMapFactory {
private static final LogService LOGGER = LogServiceFactory.getLogService(
MinMaxIndexDataMapFactory.class.getName());
+ private DataMapSchema dataMapSchema;
private DataMapMeta dataMapMeta;
private String dataMapName;
private AbsoluteTableIdentifier identifier;
+ public MinMaxIndexDataMapFactory(CarbonTable carbonTable) {
+ super(carbonTable);
+ }
+
// this is an example for datamap, we can choose the columns and operations that
// will be supported by this datamap. Furthermore, we can add cache-support for this datamap.
- @Override public void init(CarbonTable carbonTable, DataMapSchema dataMapSchema)
+ @Override public void init(DataMapSchema dataMapSchema)
throws IOException, MalformedDataMapCommandException {
+ this.dataMapSchema = dataMapSchema;
this.identifier = carbonTable.getAbsoluteTableIdentifier();
this.dataMapName = dataMapSchema.getDataMapName();
// columns that will be indexed
List<CarbonColumn> allColumns = carbonTable.getCreateOrderColumn(identifier.getTableName());
- List<String> minMaxCols = (List) CollectionUtils.collect(allColumns, new Transformer() {
- @Override public Object transform(Object o) {
- return ((CarbonColumn) o).getColName();
- }
- });
- LOGGER.info("MinMaxDataMap support index columns: " + StringUtils.join(minMaxCols, ", "));
// operations that will be supported on the indexed columns
List<ExpressionType> optOperations = new ArrayList<>();
@@ -80,17 +81,24 @@ public class MinMaxIndexDataMapFactory extends CoarseGrainDataMapFactory {
optOperations.add(ExpressionType.LESSTHAN_EQUALTO);
optOperations.add(ExpressionType.NOT_EQUALS);
LOGGER.error("MinMaxDataMap support operations: " + StringUtils.join(optOperations, ", "));
- this.dataMapMeta = new DataMapMeta(minMaxCols, optOperations);
+ this.dataMapMeta = new DataMapMeta(allColumns, optOperations);
}
/**
* createWriter will return the MinMaxDataWriter.
*
* @param segment
+ * @param shardName
* @return
*/
- @Override public DataMapWriter createWriter(Segment segment, String writeDirectoryPath) {
- return new MinMaxDataWriter(identifier, dataMapName, segment, writeDirectoryPath);
+ @Override public DataMapWriter createWriter(Segment segment, String shardName) {
+ return new MinMaxDataWriter(carbonTable, dataMapSchema, segment, shardName,
+ dataMapMeta.getIndexedColumns());
+ }
+
+ @Override public DataMapRefresher createRefresher(Segment segment, String shardName)
+ throws IOException {
+ return null;
}
/**
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneCoarseGrainDataMapFactory.java
----------------------------------------------------------------------
diff --git a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneCoarseGrainDataMapFactory.java b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneCoarseGrainDataMapFactory.java
index ba38319..dca5c90 100644
--- a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneCoarseGrainDataMapFactory.java
+++ b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneCoarseGrainDataMapFactory.java
@@ -22,15 +22,19 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
import org.apache.carbondata.common.logging.LogService;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.datamap.DataMapDistributable;
import org.apache.carbondata.core.datamap.DataMapLevel;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.dev.DataMapModel;
+import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap;
import org.apache.carbondata.core.features.TableOperation;
import org.apache.carbondata.core.memory.MemoryException;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
/**
* FG level of lucene DataMap
@@ -40,6 +44,11 @@ public class LuceneCoarseGrainDataMapFactory extends LuceneDataMapFactoryBase<Co
private static final LogService LOGGER =
LogServiceFactory.getLogService(LuceneCoarseGrainDataMapFactory.class.getName());
+ public LuceneCoarseGrainDataMapFactory(CarbonTable carbonTable, DataMapSchema dataMapSchema)
+ throws MalformedDataMapCommandException {
+ super(carbonTable, dataMapSchema);
+ }
+
/**
* Get the datamap for segmentid
*/
@@ -49,7 +58,7 @@ public class LuceneCoarseGrainDataMapFactory extends LuceneDataMapFactoryBase<Co
CoarseGrainDataMap dataMap = new LuceneCoarseGrainDataMap(analyzer);
try {
dataMap.init(new DataMapModel(
- LuceneDataMapWriter.genDataMapStorePath(
+ DataMapWriter.getDefaultDataMapPath(
tableIdentifier.getTablePath(), segment.getSegmentNo(), dataMapName)));
} catch (MemoryException e) {
LOGGER.error("failed to get lucene datamap , detail is {}" + e.getMessage());
@@ -69,7 +78,7 @@ public class LuceneCoarseGrainDataMapFactory extends LuceneDataMapFactoryBase<Co
}
@Override
- public DataMapLevel getDataMapType() {
+ public DataMapLevel getDataMapLevel() {
return DataMapLevel.CG;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapFactoryBase.java
----------------------------------------------------------------------
diff --git a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapFactoryBase.java b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapFactoryBase.java
index 1cde0c1..d52cef9 100644
--- a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapFactoryBase.java
+++ b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapFactoryBase.java
@@ -35,6 +35,7 @@ import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.TableDataMap;
import org.apache.carbondata.core.datamap.dev.DataMap;
import org.apache.carbondata.core.datamap.dev.DataMapFactory;
+import org.apache.carbondata.core.datamap.dev.DataMapRefresher;
import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
@@ -51,7 +52,6 @@ import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.events.Event;
-import org.apache.commons.lang.StringUtils;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
@@ -59,9 +59,7 @@ import org.apache.lucene.analysis.standard.StandardAnalyzer;
* Base implementation for CG and FG lucene DataMapFactory.
*/
@InterfaceAudience.Internal
-abstract class LuceneDataMapFactoryBase<T extends DataMap> implements DataMapFactory<T> {
-
- static final String TEXT_COLUMNS = "text_columns";
+abstract class LuceneDataMapFactoryBase<T extends DataMap> extends DataMapFactory<T> {
/**
* Logger
@@ -88,26 +86,17 @@ abstract class LuceneDataMapFactoryBase<T extends DataMap> implements DataMapFac
*/
AbsoluteTableIdentifier tableIdentifier = null;
- /**
- * indexed carbon columns for lucene
- */
- List<String> indexedCarbonColumns = null;
-
- CarbonTable carbonTable = null;
-
-
- @Override
- public void init(CarbonTable carbonTable, DataMapSchema dataMapSchema)
- throws IOException, MalformedDataMapCommandException {
+ public LuceneDataMapFactoryBase(CarbonTable carbonTable, DataMapSchema dataMapSchema)
+ throws MalformedDataMapCommandException {
+ super(carbonTable, dataMapSchema);
Objects.requireNonNull(carbonTable.getAbsoluteTableIdentifier());
Objects.requireNonNull(dataMapSchema);
- this.carbonTable = carbonTable;
this.tableIdentifier = carbonTable.getAbsoluteTableIdentifier();
this.dataMapName = dataMapSchema.getDataMapName();
// validate DataMapSchema and get index columns
- List<String> indexedColumns = validateAndGetIndexedColumns(dataMapSchema, carbonTable);
+ List<CarbonColumn> indexedColumns = carbonTable.getIndexedColumns(dataMapSchema);
// add optimizedOperations
List<ExpressionType> optimizedOperations = new ArrayList<ExpressionType>();
@@ -126,55 +115,6 @@ abstract class LuceneDataMapFactoryBase<T extends DataMap> implements DataMapFac
}
/**
- * validate Lucene DataMap
- * 1. require TEXT_COLUMNS property
- * 2. TEXT_COLUMNS can't contains illegal argument(empty, blank)
- * 3. TEXT_COLUMNS can't contains duplicate same columns
- * 4. TEXT_COLUMNS should be exists in table columns
- * 5. TEXT_COLUMNS support only String DataType columns
- */
- public static List<String> validateAndGetIndexedColumns(DataMapSchema dataMapSchema,
- CarbonTable carbonTable) throws MalformedDataMapCommandException {
- String textColumnsStr = dataMapSchema.getProperties().get(TEXT_COLUMNS);
- if (textColumnsStr == null || StringUtils.isBlank(textColumnsStr)) {
- throw new MalformedDataMapCommandException(
- "Lucene DataMap require proper TEXT_COLUMNS property.");
- }
- String[] textColumns = textColumnsStr.split(",", -1);
- for (int i = 0; i < textColumns.length; i++) {
- textColumns[i] = textColumns[i].trim().toLowerCase();
- }
- for (int i = 0; i < textColumns.length; i++) {
- if (textColumns[i].isEmpty()) {
- throw new MalformedDataMapCommandException("TEXT_COLUMNS contains illegal argument.");
- }
- for (int j = i + 1; j < textColumns.length; j++) {
- if (textColumns[i].equals(textColumns[j])) {
- throw new MalformedDataMapCommandException(
- "TEXT_COLUMNS has duplicate columns :" + textColumns[i]);
- }
- }
- }
- List<String> indexedCarbonColumns = new ArrayList<>(textColumns.length);
- for (int i = 0; i < textColumns.length; i++) {
- CarbonColumn column = carbonTable.getColumnByName(carbonTable.getTableName(), textColumns[i]);
- if (null == column) {
- throw new MalformedDataMapCommandException("TEXT_COLUMNS: " + textColumns[i]
- + " does not exist in table. Please check create DataMap statement.");
- } else if (column.getDataType() != DataTypes.STRING) {
- throw new MalformedDataMapCommandException(
- "TEXT_COLUMNS only supports String column. " + "Unsupported column: " + textColumns[i]
- + ", DataType: " + column.getDataType());
- } else if (column.getEncoder().contains(Encoding.DICTIONARY)) {
- throw new MalformedDataMapCommandException(
- "TEXT_COLUMNS cannot contain dictionary column " + column.getColName());
- }
- indexedCarbonColumns.add(column.getColName());
- }
- return indexedCarbonColumns;
- }
-
- /**
* this method will delete the datamap folders during drop datamap
* @throws MalformedDataMapCommandException
*/
@@ -206,10 +146,16 @@ abstract class LuceneDataMapFactoryBase<T extends DataMap> implements DataMapFac
* Return a new write for this datamap
*/
@Override
- public DataMapWriter createWriter(Segment segment, String writeDirectoryPath) {
- LOGGER.info("lucene data write to " + writeDirectoryPath);
- return new LuceneDataMapWriter(tableIdentifier, dataMapName, segment, writeDirectoryPath, true,
- indexedCarbonColumns);
+ public DataMapWriter createWriter(Segment segment, String shardName) {
+ LOGGER.info("lucene data write to " + shardName);
+ return new LuceneDataMapWriter(getCarbonTable().getTablePath(), dataMapName,
+ dataMapMeta.getIndexedColumns(), segment, shardName, true);
+ }
+
+ @Override
+ public DataMapRefresher createRefresher(Segment segment, String shardName) {
+ return new LuceneDataMapRefresher(getCarbonTable().getTablePath(), dataMapName,
+ segment, shardName, dataMapMeta.getIndexedColumns());
}
/**
@@ -280,7 +226,7 @@ abstract class LuceneDataMapFactoryBase<T extends DataMap> implements DataMapFac
try {
// there can be multiple lucene datamaps present on a table, so get all datamaps and form
// the path till the index file directories in all datamaps folders present in each segment
- dataMaps = DataMapStoreManager.getInstance().getAllDataMap(carbonTable);
+ dataMaps = DataMapStoreManager.getInstance().getAllDataMap(getCarbonTable());
} catch (IOException ex) {
LOGGER.error("failed to get datamaps");
}
@@ -302,4 +248,26 @@ abstract class LuceneDataMapFactoryBase<T extends DataMap> implements DataMapFac
}
return indexDirs.toArray(new CarbonFile[0]);
}
+
+ /**
+ * Further validate whether it is string column and dictionary column.
+ * Currently only string and non-dictionary column is supported for Lucene DataMap
+ */
+ @Override
+ public void validate() throws MalformedDataMapCommandException {
+ super.validate();
+ List<CarbonColumn> indexColumns = getCarbonTable().getIndexedColumns(getDataMapSchema());
+
+ for (CarbonColumn column : indexColumns) {
+ if (column.getDataType() != DataTypes.STRING) {
+ throw new MalformedDataMapCommandException(String.format(
+ "Only String column is supported, column '%s' is %s type. ",
+ column.getColName(), column.getDataType()));
+ } else if (column.getEncoder().contains(Encoding.DICTIONARY)) {
+ throw new MalformedDataMapCommandException(String.format(
+ "Dictionary column is not supported, column '%s' is dictionary column",
+ column.getColName()));
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapRefresher.java
----------------------------------------------------------------------
diff --git a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapRefresher.java b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapRefresher.java
new file mode 100644
index 0000000..ee500ef
--- /dev/null
+++ b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapRefresher.java
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.datamap.lucene;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datamap.Segment;
+import org.apache.carbondata.core.datamap.dev.DataMapRefresher;
+import org.apache.carbondata.core.datastore.impl.FileFactory;
+import org.apache.carbondata.core.metadata.datatype.DataType;
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
+import org.apache.carbondata.core.util.CarbonProperties;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.lucene.analysis.Analyzer;
+import org.apache.lucene.analysis.standard.StandardAnalyzer;
+import org.apache.lucene.codecs.lucene50.Lucene50StoredFieldsFormat;
+import org.apache.lucene.codecs.lucene62.Lucene62Codec;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.DoublePoint;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.FloatPoint;
+import org.apache.lucene.document.IntPoint;
+import org.apache.lucene.document.IntRangeField;
+import org.apache.lucene.document.LongPoint;
+import org.apache.lucene.document.StoredField;
+import org.apache.lucene.document.TextField;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.RAMDirectory;
+import org.apache.solr.store.hdfs.HdfsDirectory;
+
+public class LuceneDataMapRefresher implements DataMapRefresher {
+
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(LuceneDataMapWriter.class.getName());
+
+ private String dataMapPath;
+
+ private List<CarbonColumn> indexColumns;
+
+ private int columnsCount;
+
+ private IndexWriter indexWriter = null;
+
+ private IndexWriter pageIndexWriter = null;
+
+ private Analyzer analyzer = null;
+
+ LuceneDataMapRefresher(String tablePath, String dataMapName,
+ Segment segment, String shardName, List<CarbonColumn> indexColumns) {
+ this.dataMapPath = CarbonTablePath.getDataMapStorePathOnShardName(
+ tablePath, segment.getSegmentNo(), dataMapName, shardName);
+ this.indexColumns = indexColumns;
+ this.columnsCount = indexColumns.size();
+ }
+
+ @Override
+ public void initialize() throws IOException {
+ // get index path, put index data into segment's path
+ Path indexPath = FileFactory.getPath(dataMapPath);
+ FileSystem fs = FileFactory.getFileSystem(indexPath);
+
+ // if index path exists, should delete it because we are
+ // rebuilding the whole datamap for all segments
+ if (fs.exists(indexPath)) {
+ fs.delete(indexPath, true);
+ }
+ if (!fs.mkdirs(indexPath)) {
+ LOGGER.error("Failed to create directory " + indexPath);
+ }
+
+ if (null == analyzer) {
+ analyzer = new StandardAnalyzer();
+ }
+
+ // create a index writer
+ Directory indexDir = new HdfsDirectory(indexPath, FileFactory.getConfiguration());
+
+ IndexWriterConfig indexWriterConfig = new IndexWriterConfig(analyzer);
+ if (CarbonProperties.getInstance()
+ .getProperty(CarbonCommonConstants.CARBON_LUCENE_COMPRESSION_MODE,
+ CarbonCommonConstants.CARBON_LUCENE_COMPRESSION_MODE_DEFAULT)
+ .equalsIgnoreCase(CarbonCommonConstants.CARBON_LUCENE_COMPRESSION_MODE_DEFAULT)) {
+ indexWriterConfig.setCodec(new Lucene62Codec(Lucene50StoredFieldsFormat.Mode.BEST_SPEED));
+ } else {
+ indexWriterConfig
+ .setCodec(new Lucene62Codec(Lucene50StoredFieldsFormat.Mode.BEST_COMPRESSION));
+ }
+
+ indexWriter = new IndexWriter(indexDir, new IndexWriterConfig(analyzer));
+ }
+
+ private IndexWriter createPageIndexWriter() throws IOException {
+ // save index data into ram, write into disk after one page finished
+ RAMDirectory ramDir = new RAMDirectory();
+ return new IndexWriter(ramDir, new IndexWriterConfig(analyzer));
+ }
+
+ private void addPageIndex(IndexWriter pageIndexWriter) throws IOException {
+
+ Directory directory = pageIndexWriter.getDirectory();
+
+ // close ram writer
+ pageIndexWriter.close();
+
+ // add ram index data into disk
+ indexWriter.addIndexes(directory);
+
+ // delete this ram data
+ directory.close();
+ }
+
+ @Override
+ public void addRow(int blockletId, int pageId, int rowId, Object[] values) throws IOException {
+ if (rowId == 0) {
+ if (pageIndexWriter != null) {
+ addPageIndex(pageIndexWriter);
+ }
+ pageIndexWriter = createPageIndexWriter();
+ }
+
+ // create a new document
+ Document doc = new Document();
+
+ // add blocklet Id
+ doc.add(new IntPoint(LuceneDataMapWriter.BLOCKLETID_NAME, (int) values[columnsCount]));
+ doc.add(new StoredField(LuceneDataMapWriter.BLOCKLETID_NAME, (int) values[columnsCount]));
+
+ // add page id
+ doc.add(new IntPoint(LuceneDataMapWriter.PAGEID_NAME, (int) values[columnsCount + 1]));
+ doc.add(new StoredField(LuceneDataMapWriter.PAGEID_NAME, (int) values[columnsCount + 1]));
+
+ // add row id
+ doc.add(new IntPoint(LuceneDataMapWriter.ROWID_NAME, rowId));
+ doc.add(new StoredField(LuceneDataMapWriter.ROWID_NAME, rowId));
+
+ // add other fields
+ for (int colIdx = 0; colIdx < columnsCount; colIdx++) {
+ CarbonColumn column = indexColumns.get(colIdx);
+ addField(doc, column.getColName(), column.getDataType(), values[colIdx]);
+ }
+
+ pageIndexWriter.addDocument(doc);
+ }
+
+ private boolean addField(Document doc, String fieldName, DataType type, Object value) {
+ if (type == DataTypes.STRING) {
+ doc.add(new TextField(fieldName, (String) value, Field.Store.NO));
+ } else if (type == DataTypes.BYTE) {
+ // byte type , use int range to deal with byte, lucene has no byte type
+ IntRangeField field =
+ new IntRangeField(fieldName, new int[] { Byte.MIN_VALUE }, new int[] { Byte.MAX_VALUE });
+ field.setIntValue((int) value);
+ doc.add(field);
+ } else if (type == DataTypes.SHORT) {
+ // short type , use int range to deal with short type, lucene has no short type
+ IntRangeField field = new IntRangeField(fieldName, new int[] { Short.MIN_VALUE },
+ new int[] { Short.MAX_VALUE });
+ field.setShortValue((short) value);
+ doc.add(field);
+ } else if (type == DataTypes.INT) {
+ // int type , use int point to deal with int type
+ doc.add(new IntPoint(fieldName, (int) value));
+ } else if (type == DataTypes.LONG) {
+ // long type , use long point to deal with long type
+ doc.add(new LongPoint(fieldName, (long) value));
+ } else if (type == DataTypes.FLOAT) {
+ doc.add(new FloatPoint(fieldName, (float) value));
+ } else if (type == DataTypes.DOUBLE) {
+ doc.add(new DoublePoint(fieldName, (double) value));
+ } else if (type == DataTypes.DATE) {
+ // TODO: how to get data value
+ } else if (type == DataTypes.TIMESTAMP) {
+ // TODO: how to get
+ } else if (type == DataTypes.BOOLEAN) {
+ IntRangeField field = new IntRangeField(fieldName, new int[] { 0 }, new int[] { 1 });
+ field.setIntValue((boolean) value ? 1 : 0);
+ doc.add(field);
+ } else {
+ LOGGER.error("unsupport data type " + type);
+ throw new RuntimeException("unsupported data type " + type);
+ }
+ return true;
+ }
+
+ @Override
+ public void finish() throws IOException {
+ if (indexWriter != null && pageIndexWriter != null) {
+ addPageIndex(pageIndexWriter);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (indexWriter != null) {
+ indexWriter.close();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapWriter.java
----------------------------------------------------------------------
diff --git a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapWriter.java b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapWriter.java
index 95823bb..12fa1ef 100644
--- a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapWriter.java
+++ b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapWriter.java
@@ -17,7 +17,6 @@
package org.apache.carbondata.datamap.lucene;
-import java.io.File;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.List;
@@ -30,11 +29,10 @@ import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.datastore.page.ColumnPage;
-import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.util.CarbonProperties;
-import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -76,12 +74,8 @@ public class LuceneDataMapWriter extends DataMapWriter {
private Analyzer analyzer = null;
- private String dataMapName = null;
-
private boolean isFineGrain = true;
- private List<String> indexedCarbonColumns = null;
-
public static final String BLOCKLETID_NAME = "blockletId";
private String indexShardName = null;
@@ -90,69 +84,54 @@ public class LuceneDataMapWriter extends DataMapWriter {
public static final String ROWID_NAME = "rowId";
- LuceneDataMapWriter(AbsoluteTableIdentifier identifier, String dataMapName, Segment segment,
- String writeDirectoryPath, boolean isFineGrain, List<String> indexedCarbonColumns) {
- super(identifier, segment, writeDirectoryPath);
- this.dataMapName = dataMapName;
+ LuceneDataMapWriter(String tablePath, String dataMapName, List<CarbonColumn> indexColumns,
+ Segment segment, String shardName, boolean isFineGrain) {
+ super(tablePath, dataMapName, indexColumns, segment, shardName);
this.isFineGrain = isFineGrain;
- this.indexedCarbonColumns = indexedCarbonColumns;
- }
-
- private String getIndexPath(String taskName) {
- if (isFineGrain) {
- return genDataMapStorePathOnTaskId(identifier.getTablePath(), segmentId, dataMapName,
- taskName);
- } else {
- // TODO: where write data in coarse grain data map
- return genDataMapStorePathOnTaskId(identifier.getTablePath(), segmentId, dataMapName,
- taskName);
- }
}
/**
* Start of new block notification.
*/
- public void onBlockStart(String blockId, String indexShardName) throws IOException {
- if (this.indexShardName == null || !this.indexShardName.equals(indexShardName)) {
- if (indexWriter != null) {
- return;
- }
- // get index path, put index data into segment's path
- String strIndexPath = getIndexPath(indexShardName);
- Path indexPath = FileFactory.getPath(strIndexPath);
- FileSystem fs = FileFactory.getFileSystem(indexPath);
-
- // if index path not exists, create it
- if (!fs.exists(indexPath)) {
- fs.mkdirs(indexPath);
+ public void onBlockStart(String blockId) throws IOException {
+ if (indexWriter != null) {
+ return;
+ }
+ // get index path, put index data into segment's path
+ Path indexPath = FileFactory.getPath(dataMapPath);
+ FileSystem fs = FileFactory.getFileSystem(indexPath);
+
+ // if index path not exists, create it
+ if (!fs.exists(indexPath)) {
+ if (!fs.mkdirs(indexPath)) {
+ throw new IOException("Failed to create directory " + dataMapPath);
}
+ }
- if (null == analyzer) {
- analyzer = new StandardAnalyzer();
- }
+ if (null == analyzer) {
+ analyzer = new StandardAnalyzer();
+ }
- // the indexWriter closes the FileSystem on closing the writer, so for a new configuration
- // and disable the cache for the index writer, it will be closed on closing the writer
- Configuration conf = new Configuration();
- conf.set("fs.hdfs.impl.disable.cache", "true");
-
- // create a index writer
- Directory indexDir = new HdfsDirectory(indexPath, conf);
-
- IndexWriterConfig indexWriterConfig = new IndexWriterConfig(analyzer);
- if (CarbonProperties.getInstance()
- .getProperty(CarbonCommonConstants.CARBON_LUCENE_COMPRESSION_MODE,
- CarbonCommonConstants.CARBON_LUCENE_COMPRESSION_MODE_DEFAULT)
- .equalsIgnoreCase(CarbonCommonConstants.CARBON_LUCENE_COMPRESSION_MODE_DEFAULT)) {
- indexWriterConfig.setCodec(new Lucene62Codec(Lucene50StoredFieldsFormat.Mode.BEST_SPEED));
- } else {
- indexWriterConfig
- .setCodec(new Lucene62Codec(Lucene50StoredFieldsFormat.Mode.BEST_COMPRESSION));
- }
+ // the indexWriter closes the FileSystem on closing the writer, so for a new configuration
+ // and disable the cache for the index writer, it will be closed on closing the writer
+ Configuration conf = new Configuration();
+ conf.set("fs.hdfs.impl.disable.cache", "true");
+
+ // create a index writer
+ Directory indexDir = new HdfsDirectory(indexPath, conf);
- indexWriter = new IndexWriter(indexDir, new IndexWriterConfig(analyzer));
+ IndexWriterConfig indexWriterConfig = new IndexWriterConfig(analyzer);
+ if (CarbonProperties.getInstance()
+ .getProperty(CarbonCommonConstants.CARBON_LUCENE_COMPRESSION_MODE,
+ CarbonCommonConstants.CARBON_LUCENE_COMPRESSION_MODE_DEFAULT)
+ .equalsIgnoreCase(CarbonCommonConstants.CARBON_LUCENE_COMPRESSION_MODE_DEFAULT)) {
+ indexWriterConfig.setCodec(new Lucene62Codec(Lucene50StoredFieldsFormat.Mode.BEST_SPEED));
+ } else {
+ indexWriterConfig
+ .setCodec(new Lucene62Codec(Lucene50StoredFieldsFormat.Mode.BEST_COMPRESSION));
}
+ indexWriter = new IndexWriter(indexDir, new IndexWriterConfig(analyzer));
}
/**
@@ -162,18 +141,30 @@ public class LuceneDataMapWriter extends DataMapWriter {
}
+ private RAMDirectory ramDir;
+ private IndexWriter ramIndexWriter;
+
/**
* Start of new blocklet notification.
*/
- public void onBlockletStart(int blockletId) {
-
+ public void onBlockletStart(int blockletId) throws IOException {
+ // save index data into ram, write into disk after one page finished
+ ramDir = new RAMDirectory();
+ ramIndexWriter = new IndexWriter(ramDir, new IndexWriterConfig(analyzer));
}
/**
* End of blocklet notification
*/
- public void onBlockletEnd(int blockletId) {
+ public void onBlockletEnd(int blockletId) throws IOException {
+ // close ram writer
+ ramIndexWriter.close();
+
+ // add ram index data into disk
+ indexWriter.addIndexes(ramDir);
+ // delete this ram data
+ ramDir.close();
}
/**
@@ -182,72 +173,54 @@ public class LuceneDataMapWriter extends DataMapWriter {
* Implementation should copy the content of `pages` as needed, because `pages` memory
* may be freed after this method returns, if using unsafe column page.
*/
- public void onPageAdded(int blockletId, int pageId, ColumnPage[] pages) throws IOException {
- // save index data into ram, write into disk after one page finished
- RAMDirectory ramDir = new RAMDirectory();
- IndexWriter ramIndexWriter = new IndexWriter(ramDir, new IndexWriterConfig(analyzer));
-
- int columnsCount = pages.length;
- if (columnsCount <= 0) {
- LOGGER.warn("empty data");
- ramIndexWriter.close();
- ramDir.close();
- return;
- }
-
- int pageSize = pages[0].getPageSize();
+ public void onPageAdded(int blockletId, int pageId, int pageSize, ColumnPage[] pages)
+ throws IOException {
for (int rowId = 0; rowId < pageSize; rowId++) {
// create a new document
Document doc = new Document();
// add blocklet Id
- doc.add(new IntPoint(BLOCKLETID_NAME, new int[] { blockletId }));
+ doc.add(new IntPoint(BLOCKLETID_NAME, blockletId));
doc.add(new StoredField(BLOCKLETID_NAME, blockletId));
//doc.add(new NumericDocValuesField(BLOCKLETID_NAME,blockletId));
// add page id and row id in Fine Grain data map
if (isFineGrain) {
// add page Id
- doc.add(new IntPoint(PAGEID_NAME, new int[] { pageId }));
+ doc.add(new IntPoint(PAGEID_NAME, pageId));
doc.add(new StoredField(PAGEID_NAME, pageId));
//doc.add(new NumericDocValuesField(PAGEID_NAME,pageId));
// add row id
- doc.add(new IntPoint(ROWID_NAME, new int[] { rowId }));
+ doc.add(new IntPoint(ROWID_NAME, rowId));
doc.add(new StoredField(ROWID_NAME, rowId));
//doc.add(new NumericDocValuesField(ROWID_NAME,rowId));
}
// add indexed columns value into the document
- for (ColumnPage page : pages) {
- if (!page.getNullBits().get(rowId)) {
- addField(doc, page, rowId, Field.Store.NO);
+ List<CarbonColumn> indexColumns = getIndexColumns();
+ for (int i = 0; i < pages.length; i++) {
+ // add to lucene only if value is not null
+ if (!pages[i].getNullBits().get(rowId)) {
+ addField(doc, pages[i].getData(rowId), indexColumns.get(i), Field.Store.NO);
}
}
// add this document
ramIndexWriter.addDocument(doc);
-
}
- // close ram writer
- ramIndexWriter.close();
-
- // add ram index data into disk
- indexWriter.addIndexes(new Directory[] { ramDir });
- // delete this ram data
- ramDir.close();
}
- private boolean addField(Document doc, ColumnPage page, int rowId, Field.Store store) {
+ private boolean addField(Document doc, Object data, CarbonColumn column, Field.Store store) {
//get field name
- String fieldName = page.getColumnSpec().getFieldName();
+ String fieldName = column.getColName();
//get field type
- DataType type = page.getColumnSpec().getSchemaDataType();
+ DataType type = column.getDataType();
if (type == DataTypes.BYTE) {
// byte type , use int range to deal with byte, lucene has no byte type
- byte value = page.getByte(rowId);
+ byte value = (byte) data;
IntRangeField field =
new IntRangeField(fieldName, new int[] { Byte.MIN_VALUE }, new int[] { Byte.MAX_VALUE });
field.setIntValue(value);
@@ -259,7 +232,7 @@ public class LuceneDataMapWriter extends DataMapWriter {
}
} else if (type == DataTypes.SHORT) {
// short type , use int range to deal with short type, lucene has no short type
- short value = page.getShort(rowId);
+ short value = (short) data;
IntRangeField field = new IntRangeField(fieldName, new int[] { Short.MIN_VALUE },
new int[] { Short.MAX_VALUE });
field.setShortValue(value);
@@ -271,8 +244,8 @@ public class LuceneDataMapWriter extends DataMapWriter {
}
} else if (type == DataTypes.INT) {
// int type , use int point to deal with int type
- int value = page.getInt(rowId);
- doc.add(new IntPoint(fieldName, new int[] { value }));
+ int value = (int) data;
+ doc.add(new IntPoint(fieldName, value));
// if need store it , add StoredField
if (store == Field.Store.YES) {
@@ -280,27 +253,27 @@ public class LuceneDataMapWriter extends DataMapWriter {
}
} else if (type == DataTypes.LONG) {
// long type , use long point to deal with long type
- long value = page.getLong(rowId);
- doc.add(new LongPoint(fieldName, new long[] { value }));
+ long value = (long) data;
+ doc.add(new LongPoint(fieldName, value));
// if need store it , add StoredField
if (store == Field.Store.YES) {
doc.add(new StoredField(fieldName, value));
}
} else if (type == DataTypes.FLOAT) {
- float value = page.getFloat(rowId);
- doc.add(new FloatPoint(fieldName, new float[] { value }));
+ float value = (float) data;
+ doc.add(new FloatPoint(fieldName, value));
if (store == Field.Store.YES) {
doc.add(new FloatPoint(fieldName, value));
}
} else if (type == DataTypes.DOUBLE) {
- double value = page.getDouble(rowId);
- doc.add(new DoublePoint(fieldName, new double[] { value }));
+ double value = (double) data;
+ doc.add(new DoublePoint(fieldName, value));
if (store == Field.Store.YES) {
doc.add(new DoublePoint(fieldName, value));
}
} else if (type == DataTypes.STRING) {
- byte[] value = page.getBytes(rowId);
+ byte[] value = (byte[]) data;
// TODO: how to get string value
String strValue = null;
try {
@@ -310,11 +283,11 @@ public class LuceneDataMapWriter extends DataMapWriter {
}
doc.add(new TextField(fieldName, strValue, store));
} else if (type == DataTypes.DATE) {
- // TODO: how to get data value
+ throw new RuntimeException("unsupported data type " + type);
} else if (type == DataTypes.TIMESTAMP) {
- // TODO: how to get
+ throw new RuntimeException("unsupported data type " + type);
} else if (type == DataTypes.BOOLEAN) {
- boolean value = page.getBoolean(rowId);
+ boolean value = (boolean) data;
IntRangeField field = new IntRangeField(fieldName, new int[] { 0 }, new int[] { 1 });
field.setIntValue(value ? 1 : 0);
doc.add(field);
@@ -339,23 +312,4 @@ public class LuceneDataMapWriter extends DataMapWriter {
}
}
- /**
- * Return store path for datamap
- */
- public static String genDataMapStorePath(String tablePath, String segmentId, String dataMapName) {
- return CarbonTablePath.getSegmentPath(tablePath, segmentId) + File.separator + dataMapName;
- }
-
- /**
- * Return store path for datamap based on the taskName,if three tasks get launched during loading,
- * then three folders will be created based on the three task Ids and lucene index file will be
- * written into those folders
- *
- * @return store path based on taskID
- */
- public static String genDataMapStorePathOnTaskId(String tablePath, String segmentId,
- String dataMapName, String taskName) {
- return CarbonTablePath.getSegmentPath(tablePath, segmentId) + File.separator + dataMapName
- + File.separator + taskName;
- }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMap.java
----------------------------------------------------------------------
diff --git a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMap.java b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMap.java
index 1104d09..86fba32 100644
--- a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMap.java
+++ b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMap.java
@@ -87,7 +87,7 @@ public class LuceneFineGrainDataMap extends FineGrainDataMap {
*/
private Analyzer analyzer;
- private String taskName;
+ private String filePath;
LuceneFineGrainDataMap(Analyzer analyzer) {
this.analyzer = analyzer;
@@ -102,7 +102,7 @@ public class LuceneFineGrainDataMap extends FineGrainDataMap {
LOGGER.info("Lucene index read path " + indexPath.toString());
- this.taskName = indexPath.getName();
+ this.filePath = indexPath.getName();
// get file system , use hdfs file system , realized in solr project
FileSystem fs = FileFactory.getFileSystem(indexPath);
@@ -277,7 +277,7 @@ public class LuceneFineGrainDataMap extends FineGrainDataMap {
}
// add a FineGrainBlocklet
- blocklets.add(new FineGrainBlocklet(taskName, blockletId, pages));
+ blocklets.add(new FineGrainBlocklet(filePath, blockletId, pages));
}
return blocklets;
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMapFactory.java
----------------------------------------------------------------------
diff --git a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMapFactory.java b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMapFactory.java
index 1dae9b5..ec9283d 100644
--- a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMapFactory.java
+++ b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneFineGrainDataMapFactory.java
@@ -22,13 +22,17 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
import org.apache.carbondata.core.datamap.DataMapDistributable;
import org.apache.carbondata.core.datamap.DataMapLevel;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.dev.DataMapModel;
+import org.apache.carbondata.core.datamap.dev.DataMapWriter;
import org.apache.carbondata.core.datamap.dev.fgdatamap.FineGrainDataMap;
import org.apache.carbondata.core.features.TableOperation;
import org.apache.carbondata.core.memory.MemoryException;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
/**
* CG level of lucene DataMap
@@ -36,6 +40,11 @@ import org.apache.carbondata.core.memory.MemoryException;
@InterfaceAudience.Internal
public class LuceneFineGrainDataMapFactory extends LuceneDataMapFactoryBase<FineGrainDataMap> {
+ public LuceneFineGrainDataMapFactory(CarbonTable carbonTable, DataMapSchema dataMapSchema)
+ throws MalformedDataMapCommandException {
+ super(carbonTable, dataMapSchema);
+ }
+
/**
* Get the datamap for segmentid
*/
@@ -44,7 +53,7 @@ public class LuceneFineGrainDataMapFactory extends LuceneDataMapFactoryBase<Fine
FineGrainDataMap dataMap = new LuceneFineGrainDataMap(analyzer);
try {
dataMap.init(new DataMapModel(
- LuceneDataMapWriter.genDataMapStorePath(
+ DataMapWriter.getDefaultDataMapPath(
tableIdentifier.getTablePath(), segment.getSegmentNo(), dataMapName)));
} catch (MemoryException e) {
LOGGER.error("failed to get lucene datamap , detail is {}" + e.getMessage());
@@ -74,7 +83,7 @@ public class LuceneFineGrainDataMapFactory extends LuceneDataMapFactoryBase<Fine
}
@Override
- public DataMapLevel getDataMapType() {
+ public DataMapLevel getDataMapLevel() {
return DataMapLevel.FG;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneIndexRefreshBuilder.java
----------------------------------------------------------------------
diff --git a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneIndexRefreshBuilder.java b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneIndexRefreshBuilder.java
deleted file mode 100644
index 2d5dcf8..0000000
--- a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneIndexRefreshBuilder.java
+++ /dev/null
@@ -1,220 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.carbondata.datamap.lucene;
-
-import java.io.IOException;
-
-import org.apache.carbondata.common.logging.LogService;
-import org.apache.carbondata.common.logging.LogServiceFactory;
-import org.apache.carbondata.core.constants.CarbonCommonConstants;
-import org.apache.carbondata.core.datastore.impl.FileFactory;
-import org.apache.carbondata.core.metadata.datatype.DataType;
-import org.apache.carbondata.core.metadata.datatype.DataTypes;
-import org.apache.carbondata.core.util.CarbonProperties;
-
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.lucene.analysis.Analyzer;
-import org.apache.lucene.analysis.standard.StandardAnalyzer;
-import org.apache.lucene.codecs.lucene50.Lucene50StoredFieldsFormat;
-import org.apache.lucene.codecs.lucene62.Lucene62Codec;
-import org.apache.lucene.document.Document;
-import org.apache.lucene.document.DoublePoint;
-import org.apache.lucene.document.Field;
-import org.apache.lucene.document.FloatPoint;
-import org.apache.lucene.document.IntPoint;
-import org.apache.lucene.document.IntRangeField;
-import org.apache.lucene.document.LongPoint;
-import org.apache.lucene.document.StoredField;
-import org.apache.lucene.document.TextField;
-import org.apache.lucene.index.IndexWriter;
-import org.apache.lucene.index.IndexWriterConfig;
-import org.apache.lucene.store.Directory;
-import org.apache.lucene.store.RAMDirectory;
-import org.apache.solr.store.hdfs.HdfsDirectory;
-
-public class LuceneIndexRefreshBuilder {
-
- private static final LogService LOGGER =
- LogServiceFactory.getLogService(LuceneDataMapWriter.class.getName());
-
- private String strIndexPath;
-
- private String[] indexColumns;
- private DataType[] dataTypes;
-
- private int columnsCount;
-
- private IndexWriter indexWriter = null;
-
- private IndexWriter pageIndexWriter = null;
-
- private Analyzer analyzer = null;
-
- public LuceneIndexRefreshBuilder(String strIndexPath, String[] indexColumns,
- DataType[] dataTypes) {
- this.strIndexPath = strIndexPath;
- this.indexColumns = indexColumns;
- this.columnsCount = indexColumns.length;
- this.dataTypes = dataTypes;
- }
-
- public void initialize() throws IOException {
- // get index path, put index data into segment's path
- Path indexPath = FileFactory.getPath(strIndexPath);
- FileSystem fs = FileFactory.getFileSystem(indexPath);
-
- // if index path not exists, create it
- if (!fs.exists(indexPath)) {
- fs.mkdirs(indexPath);
- }
-
- if (null == analyzer) {
- analyzer = new StandardAnalyzer();
- }
-
- // create a index writer
- Directory indexDir = new HdfsDirectory(indexPath, FileFactory.getConfiguration());
-
- IndexWriterConfig indexWriterConfig = new IndexWriterConfig(analyzer);
- if (CarbonProperties.getInstance()
- .getProperty(CarbonCommonConstants.CARBON_LUCENE_COMPRESSION_MODE,
- CarbonCommonConstants.CARBON_LUCENE_COMPRESSION_MODE_DEFAULT)
- .equalsIgnoreCase(CarbonCommonConstants.CARBON_LUCENE_COMPRESSION_MODE_DEFAULT)) {
- indexWriterConfig.setCodec(new Lucene62Codec(Lucene50StoredFieldsFormat.Mode.BEST_SPEED));
- } else {
- indexWriterConfig
- .setCodec(new Lucene62Codec(Lucene50StoredFieldsFormat.Mode.BEST_COMPRESSION));
- }
-
- indexWriter = new IndexWriter(indexDir, new IndexWriterConfig(analyzer));
- }
-
- private IndexWriter createPageIndexWriter() throws IOException {
- // save index data into ram, write into disk after one page finished
- RAMDirectory ramDir = new RAMDirectory();
- IndexWriter ramIndexWriter = new IndexWriter(ramDir, new IndexWriterConfig(analyzer));
-
- return ramIndexWriter;
- }
-
- private void addPageIndex(IndexWriter pageIndexWriter) throws IOException {
-
- Directory directory = pageIndexWriter.getDirectory();
-
- // close ram writer
- pageIndexWriter.close();
-
- // add ram index data into disk
- indexWriter.addIndexes(new Directory[] { directory });
-
- // delete this ram data
- directory.close();
- }
-
- public void addDocument(Object[] values) throws IOException {
-
- if (values.length != indexColumns.length + 3) {
- throw new IOException("The column number (" + values.length + ") of the row is incorrect.");
- }
- int rowId = (int) values[indexColumns.length + 2];
- if (rowId == 0) {
- if (pageIndexWriter != null) {
- addPageIndex(pageIndexWriter);
- }
- pageIndexWriter = createPageIndexWriter();
- }
-
- // create a new document
- Document doc = new Document();
-
- // add blocklet Id
- doc.add(new IntPoint(LuceneDataMapWriter.BLOCKLETID_NAME,
- new int[] { (int) values[columnsCount] }));
- doc.add(new StoredField(LuceneDataMapWriter.BLOCKLETID_NAME, (int) values[columnsCount]));
-
- // add page id
- doc.add(new IntPoint(LuceneDataMapWriter.PAGEID_NAME,
- new int[] { (int) values[columnsCount + 1] }));
- doc.add(new StoredField(LuceneDataMapWriter.PAGEID_NAME, (int) values[columnsCount + 1]));
-
- // add row id
- doc.add(new IntPoint(LuceneDataMapWriter.ROWID_NAME, new int[] { rowId }));
- doc.add(new StoredField(LuceneDataMapWriter.ROWID_NAME, rowId));
-
- // add other fields
- for (int colIdx = 0; colIdx < columnsCount; colIdx++) {
- addField(doc, indexColumns[colIdx], dataTypes[colIdx], values[colIdx]);
- }
-
- pageIndexWriter.addDocument(doc);
- }
-
- private boolean addField(Document doc, String fieldName, DataType type, Object value) {
- if (type == DataTypes.BYTE) {
- // byte type , use int range to deal with byte, lucene has no byte type
- IntRangeField field =
- new IntRangeField(fieldName, new int[] { Byte.MIN_VALUE }, new int[] { Byte.MAX_VALUE });
- field.setIntValue((int) value);
- doc.add(field);
- } else if (type == DataTypes.SHORT) {
- // short type , use int range to deal with short type, lucene has no short type
- IntRangeField field = new IntRangeField(fieldName, new int[] { Short.MIN_VALUE },
- new int[] { Short.MAX_VALUE });
- field.setShortValue((short) value);
- doc.add(field);
- } else if (type == DataTypes.INT) {
- // int type , use int point to deal with int type
- doc.add(new IntPoint(fieldName, new int[] { (int) value }));
- } else if (type == DataTypes.LONG) {
- // long type , use long point to deal with long type
- doc.add(new LongPoint(fieldName, new long[] { (long) value }));
- } else if (type == DataTypes.FLOAT) {
- doc.add(new FloatPoint(fieldName, new float[] { (float) value }));
- } else if (type == DataTypes.DOUBLE) {
- doc.add(new DoublePoint(fieldName, new double[] { (double) value }));
- } else if (type == DataTypes.STRING) {
- doc.add(new TextField(fieldName, (String) value, Field.Store.NO));
- } else if (type == DataTypes.DATE) {
- // TODO: how to get data value
- } else if (type == DataTypes.TIMESTAMP) {
- // TODO: how to get
- } else if (type == DataTypes.BOOLEAN) {
- IntRangeField field = new IntRangeField(fieldName, new int[] { 0 }, new int[] { 1 });
- field.setIntValue((boolean) value ? 1 : 0);
- doc.add(field);
- } else {
- LOGGER.error("unsupport data type " + type);
- throw new RuntimeException("unsupported data type " + type);
- }
- return true;
- }
-
- public void finish() throws IOException {
- if (indexWriter != null && pageIndexWriter != null) {
- addPageIndex(pageIndexWriter);
- }
- }
-
- public void close() throws IOException {
- if (indexWriter != null) {
- indexWriter.close();
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonInputFormat.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonInputFormat.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonInputFormat.java
index 7c0da5e..25cc228 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonInputFormat.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonInputFormat.java
@@ -491,8 +491,8 @@ m filterExpression
if (blocklet.getSegmentId().equals(segment.getSegmentNo())) {
found = true;
// Set the pruned index file to the segment for further pruning.
- String uniqueTaskName = CarbonTablePath.getUniqueTaskName(blocklet.getTaskName());
- segment.setFilteredIndexShardName(uniqueTaskName);
+ String shardName = CarbonTablePath.getShardName(blocklet.getFilePath());
+ segment.setFilteredIndexShardName(shardName);
}
}
// Add to remove segments list if not present in pruned blocklets.
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/LuceneTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/LuceneTestCase.scala b/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/LuceneTestCase.scala
index 540c8ca..d111594 100644
--- a/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/LuceneTestCase.scala
+++ b/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/LuceneTestCase.scala
@@ -45,7 +45,7 @@ class LuceneTestCase extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP lucene_datamap ON TABLE datamap_main
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='country')
+ | DMProperties('INDEX_COLUMNS'='country')
""".stripMargin)
checkExistence(sql("show datamap on table datamap_main"), true, "lucene_datamap")
sql("drop datamap if exists lucene_datamap on table datamap_main")
@@ -64,12 +64,11 @@ class LuceneTestCase extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP lucene_datamap ON TABLE datamap_main
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='id')
+ | DMProperties('INDEX_COLUMNS'='id')
""".stripMargin)
}
assert(exception_otherdataType.getMessage
- .contains("TEXT_COLUMNS only supports String column. Unsupported column: id, " +
- "DataType: INT"))
+ .contains("Only String column is supported, column 'id' is INT type."))
}
//Create Lucene DataMap With DMProperties on MainTable and Load Data and Query
@@ -84,7 +83,7 @@ class LuceneTestCase extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP lucene_datamap ON TABLE datamap_main
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='country')
+ | DMProperties('INDEX_COLUMNS'='country')
""".stripMargin)
sql(s"LOAD DATA INPATH '$csvPath' INTO TABLE datamap_main")
@@ -105,7 +104,7 @@ class LuceneTestCase extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP lucene_datamap ON TABLE datamap_main
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='country,name,serialname')
+ | DMProperties('INDEX_COLUMNS'='country,name,serialname')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$csvPath' INTO TABLE datamap_main")
checkAnswer(sql("SELECT * FROM datamap_main WHERE TEXT_MATCH('country:ch*')"),
@@ -137,7 +136,7 @@ class LuceneTestCase extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP lucene_datamap ON TABLE datamap_main
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='country,name')
+ | DMProperties('INDEX_COLUMNS'='country,name')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$csvPath' INTO TABLE datamap_main")
sql(s"LOAD DATA LOCAL INPATH '$csvPath' INTO TABLE datamap_main")
@@ -163,7 +162,7 @@ class LuceneTestCase extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP lucene_datamap ON TABLE datamap_main
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='country')
+ | DMProperties('INDEX_COLUMNS'='country')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$csvPath' INTO TABLE datamap_main OPTIONS('header'='false'," +
s"'BAD_RECORDS_LOGGER_ENABLE'='FALSE','BAD_RECORDS_ACTION'='FORCE','SINGLE_PASS'='TRUE')")
@@ -183,7 +182,7 @@ class LuceneTestCase extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP lucene_datamap ON TABLE datamap_main
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='country,name')
+ | DMProperties('INDEX_COLUMNS'='country,name')
""".stripMargin)
sql("insert into datamap_main select 1,'abc','aa'")
sql("insert into datamap_main select 2,'def','ab'")
@@ -210,11 +209,11 @@ class LuceneTestCase extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP lucene_datamap ON TABLE datamap_main
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='country')
+ | DMProperties('INDEX_COLUMNS'='country')
""".stripMargin)
}
assert(exception_dicitionaryinclude.getMessage
- .contains("TEXT_COLUMNS cannot contain dictionary column country"))
+ .contains("Dictionary column is not supported, column 'country' is dictionary column"))
sql("drop datamap if exists lucene_datamap on table datamap_main")
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/9db662a2/integration/spark-common-test/src/test/scala/org/apache/carbondata/datamap/lucene/LuceneCoarseGrainDataMapSuite.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/datamap/lucene/LuceneCoarseGrainDataMapSuite.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/datamap/lucene/LuceneCoarseGrainDataMapSuite.scala
index 245d147..5e2534c 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/datamap/lucene/LuceneCoarseGrainDataMapSuite.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/datamap/lucene/LuceneCoarseGrainDataMapSuite.scala
@@ -55,7 +55,7 @@ class LuceneCoarseGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
s"""
| CREATE DATAMAP dm ON TABLE datamap_test
| USING 'lucene'
- | DMProperties('TEXT_COLUMNS'='name,city')
+ | DMProperties('INDEX_COLUMNS'='name,city')
""".stripMargin)
sql(s"LOAD DATA LOCAL INPATH '$file2' INTO TABLE datamap_test OPTIONS('header'='false')")