Repository: carbondata Updated Branches: refs/heads/master 289232607 -> ab9a4fc76
http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchemaCommon.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchemaCommon.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchemaCommon.scala new file mode 100644 index 0000000..cc2cc82 --- /dev/null +++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchemaCommon.scala @@ -0,0 +1,595 @@ +/* + * 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.spark.sql.execution.command + +import java.util +import java.util.UUID + +import scala.collection.JavaConverters._ +import scala.collection.mutable.Map + +import org.apache.spark.SparkContext +import org.apache.spark.sql.SQLContext +import org.apache.spark.sql.catalyst.TableIdentifier + +import org.apache.carbondata.common.logging.LogServiceFactory +import org.apache.carbondata.core.constants.CarbonCommonConstants +import org.apache.carbondata.core.metadata.CarbonTableIdentifier +import org.apache.carbondata.core.metadata.datatype.DataType +import org.apache.carbondata.core.metadata.encoder.Encoding +import org.apache.carbondata.core.metadata.schema._ +import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, TableInfo, TableSchema} +import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema +import org.apache.carbondata.core.service.CarbonCommonFactory +import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails, SegmentUpdateStatusManager} +import org.apache.carbondata.core.util.DataTypeUtil +import org.apache.carbondata.core.util.path.CarbonTablePath +import org.apache.carbondata.processing.merger.CompactionType +import org.apache.carbondata.processing.model.CarbonLoadModel +import org.apache.carbondata.spark.CarbonSparkFactory +import org.apache.carbondata.spark.load.FailureCauses +import org.apache.carbondata.spark.rdd.AlterTableAddColumnRDD +import org.apache.carbondata.spark.util.{DataTypeConverterUtil, GlobalDictionaryUtil} + +case class TableModel( + ifNotExistsSet: Boolean, + var databaseName: String, + databaseNameOp: Option[String], + tableName: String, + tableProperties: Map[String, String], + dimCols: Seq[Field], + msrCols: Seq[Field], + sortKeyDims: Option[Seq[String]], + highcardinalitydims: Option[Seq[String]], + noInvertedIdxCols: Option[Seq[String]], + columnGroups: Seq[String], + colProps: Option[util.Map[String, + util.List[ColumnProperty]]] = None, + bucketFields: Option[BucketFields], + partitionInfo: Option[PartitionInfo]) + +case class Field(column: String, var dataType: Option[String], name: Option[String], + children: Option[List[Field]], parent: String = null, + storeType: Option[String] = Some("columnar"), + var schemaOrdinal: Int = -1, + var precision: Int = 0, var scale: Int = 0, var rawSchema: String = "") + +case class ColumnProperty(key: String, value: String) + +case class ComplexField(complexType: String, primitiveField: Option[Field], + complexField: Option[ComplexField]) + +case class Partitioner(partitionClass: String, partitionColumn: Array[String], partitionCount: Int, + nodeList: Array[String]) + +case class PartitionerField(partitionColumn: String, dataType: Option[String], + columnComment: String) + +case class BucketFields(bucketColumns: Seq[String], numberOfBuckets: Int) + +case class DataLoadTableFileMapping(table: String, loadPath: String) + +case class ExecutionErrors(var failureCauses: FailureCauses, var errorMsg: String ) + +case class CarbonMergerMapping(storeLocation: String, + hdfsStoreLocation: String, + metadataFilePath: String, + var mergedLoadName: String, + databaseName: String, + factTableName: String, + validSegments: Array[String], + tableId: String, + campactionType: CompactionType, + // maxSegmentColCardinality is Cardinality of last segment of compaction + var maxSegmentColCardinality: Array[Int], + // maxSegmentColumnSchemaList is list of column schema of last segment of compaction + var maxSegmentColumnSchemaList: List[ColumnSchema]) + +case class NodeInfo(TaskId: String, noOfBlocks: Int) + +case class AlterTableModel(dbName: Option[String], + tableName: String, + segmentUpdateStatusManager: Option[SegmentUpdateStatusManager], + compactionType: String, + factTimeStamp: Option[Long], + var alterSql: String) + +case class UpdateTableModel(isUpdate: Boolean, + updatedTimeStamp: Long, + var executorErrors: ExecutionErrors) + +case class CompactionModel(compactionSize: Long, + compactionType: CompactionType, + carbonTable: CarbonTable, + isDDLTrigger: Boolean) + +case class CompactionCallableModel(storePath: String, + carbonLoadModel: CarbonLoadModel, + storeLocation: String, + carbonTable: CarbonTable, + loadsToMerge: util.List[LoadMetadataDetails], + sqlContext: SQLContext, + compactionType: CompactionType) + +case class SplitPartitionCallableModel(storePath: String, + carbonLoadModel: CarbonLoadModel, + segmentId: String, + partitionId: String, + oldPartitionIdList: List[Int], + sqlContext: SQLContext) + +case class DataTypeInfo(dataType: String, precision: Int = 0, scale: Int = 0) + +case class AlterTableDataTypeChangeModel(dataTypeInfo: DataTypeInfo, + databaseName: Option[String], + tableName: String, + columnName: String, + newColumnName: String) + +case class AlterTableRenameModel( + oldTableIdentifier: TableIdentifier, + newTableIdentifier: TableIdentifier +) + +case class AlterTableAddColumnsModel( + databaseName: Option[String], + tableName: String, + tableProperties: Map[String, String], + dimCols: Seq[Field], + msrCols: Seq[Field], + highCardinalityDims: Seq[String]) + +case class AlterTableDropColumnModel(databaseName: Option[String], + tableName: String, + columns: List[String]) + +case class AlterTableDropPartitionModel(databaseName: Option[String], + tableName: String, + partitionId: String) + +case class AlterTableSplitPartitionModel(databaseName: Option[String], + tableName: String, + partitionId: String, + splitInfo: List[String]) + +class AlterTableColumnSchemaGenerator( + alterTableModel: AlterTableAddColumnsModel, + dbName: String, + tableInfo: TableInfo, + carbonTablePath: CarbonTablePath, + tableIdentifier: CarbonTableIdentifier, + storePath: String, sc: SparkContext) { + + val LOGGER = LogServiceFactory.getLogService(this.getClass.getName) + + def process: Seq[ColumnSchema] = { + val tableSchema = tableInfo.getFactTable + val tableCols = tableSchema.getListOfColumns.asScala + val existingColsSize = tableCols.size + var allColumns = tableCols.filter(x => x.isDimensionColumn) + var newCols = Seq[ColumnSchema]() + + alterTableModel.dimCols.foreach(field => { + val encoders = new java.util.ArrayList[Encoding]() + encoders.add(Encoding.DICTIONARY) + val columnSchema: ColumnSchema = getColumnSchema( + DataTypeConverterUtil.convertToCarbonType(field.dataType.getOrElse("")), + field.name.getOrElse(field.column), + isCol = true, + encoders, + isDimensionCol = true, + -1, + field.precision, + field.scale, + field.schemaOrdinal + existingColsSize) + allColumns ++= Seq(columnSchema) + newCols ++= Seq(columnSchema) + }) + + allColumns ++= tableCols.filter(x => !x.isDimensionColumn) + alterTableModel.msrCols.foreach(field => { + val encoders = new java.util.ArrayList[Encoding]() + val columnSchema: ColumnSchema = getColumnSchema( + DataTypeConverterUtil.convertToCarbonType(field.dataType.getOrElse("")), + field.name.getOrElse(field.column), + isCol = true, + encoders, + isDimensionCol = false, + -1, + field.precision, + field.scale, + field.schemaOrdinal + existingColsSize) + allColumns ++= Seq(columnSchema) + newCols ++= Seq(columnSchema) + }) + + // Check if there is any duplicate measures or dimensions. + // Its based on the dimension name and measure name + allColumns.filter(x => !x.isInvisible).groupBy(_.getColumnName) + .foreach(f => if (f._2.size > 1) { + val name = f._1 + LOGGER.error(s"Duplicate column found with name: $name") + LOGGER.audit( + s"Validation failed for Create/Alter Table Operation " + + s"for ${ dbName }.${ alterTableModel.tableName }. " + + s"Duplicate column found with name: $name") + sys.error(s"Duplicate column found with name: $name") + }) + + val columnValidator = CarbonSparkFactory.getCarbonColumnValidator + columnValidator.validateColumns(allColumns) + + // populate table properties map + val tablePropertiesMap = tableSchema.getTableProperties + alterTableModel.tableProperties.foreach { + x => val value = tablePropertiesMap.get(x._1) + if (null != value) { + tablePropertiesMap.put(x._1, value + "," + x._2) + } else { + tablePropertiesMap.put(x._1, x._2) + } + } + // This part will create dictionary file for all newly added dictionary columns + // if valid default value is provided, + // then that value will be included while creating dictionary file + val defaultValueString = "default.value." + newCols.foreach { col => + var rawData: String = null + for (elem <- alterTableModel.tableProperties) { + if (elem._1.toLowerCase.startsWith(defaultValueString)) { + if (col.getColumnName.equalsIgnoreCase(elem._1.substring(defaultValueString.length))) { + rawData = elem._2 + val data = DataTypeUtil.convertDataToBytesBasedOnDataType(elem._2, col) + if (null != data) { + col.setDefaultValue(data) + } else { + LOGGER + .error( + "Invalid default value for new column " + dbName + "." + + alterTableModel.tableName + + "." + col.getColumnName + " : " + elem._2) + } + } + } + else if (elem._1.equalsIgnoreCase("no_inverted_index")) { + col.getEncodingList.remove(Encoding.INVERTED_INDEX) + } + } + } + tableSchema.setListOfColumns(allColumns.asJava) + tableInfo.setLastUpdatedTime(System.currentTimeMillis()) + tableInfo.setFactTable(tableSchema) + newCols + } + + private def getColumnSchema(dataType: DataType, colName: String, isCol: Boolean, + encoders: java.util.List[Encoding], isDimensionCol: Boolean, + colGroup: Integer, precision: Integer, scale: Integer, schemaOrdinal: Int): ColumnSchema = { + val columnSchema = new ColumnSchema() + columnSchema.setDataType(dataType) + columnSchema.setColumnName(colName) + if (alterTableModel.highCardinalityDims.contains(colName)) { + encoders.remove(Encoding.DICTIONARY) + } + if (dataType == DataType.TIMESTAMP || dataType == DataType.DATE) { + encoders.add(Encoding.DIRECT_DICTIONARY) + } + val colPropMap = new java.util.HashMap[String, String]() + columnSchema.setEncodingList(encoders) + val colUniqueIdGenerator = CarbonCommonFactory.getColumnUniqueIdGenerator + val columnUniqueId = colUniqueIdGenerator.generateUniqueId( + alterTableModel.databaseName.getOrElse(dbName), + columnSchema) + columnSchema.setColumnUniqueId(columnUniqueId) + columnSchema.setColumnReferenceId(columnUniqueId) + columnSchema.setColumnar(isCol) + columnSchema.setDimensionColumn(isDimensionCol) + columnSchema.setColumnGroup(colGroup) + columnSchema.setPrecision(precision) + columnSchema.setScale(scale) + columnSchema.setSchemaOrdinal(schemaOrdinal) + columnSchema.setUseInvertedIndex(isDimensionCol) + columnSchema + } +} +object TableNewProcessor { + def apply(cm: TableModel): TableInfo = { + new TableNewProcessor(cm).process + } +} + +class TableNewProcessor(cm: TableModel) { + + var index = 0 + var rowGroup = 0 + + def getAllChildren(fieldChildren: Option[List[Field]]): Seq[ColumnSchema] = { + var allColumns: Seq[ColumnSchema] = Seq[ColumnSchema]() + fieldChildren.foreach(fields => { + fields.foreach(field => { + val encoders = new java.util.ArrayList[Encoding]() + encoders.add(Encoding.DICTIONARY) + val columnSchema: ColumnSchema = getColumnSchema( + DataTypeConverterUtil.convertToCarbonType(field.dataType.getOrElse("")), + field.name.getOrElse(field.column), index, + isCol = true, encoders, isDimensionCol = true, rowGroup, field.precision, field.scale, + field.schemaOrdinal) + allColumns ++= Seq(columnSchema) + index = index + 1 + rowGroup = rowGroup + 1 + if (field.children.get != null) { + columnSchema.setNumberOfChild(field.children.get.size) + allColumns ++= getAllChildren(field.children) + } + }) + }) + allColumns + } + + def getColumnSchema(dataType: DataType, colName: String, index: Integer, isCol: Boolean, + encoders: java.util.List[Encoding], isDimensionCol: Boolean, + colGroup: Integer, precision: Integer, scale: Integer, schemaOrdinal: Int): ColumnSchema = { + val columnSchema = new ColumnSchema() + columnSchema.setDataType(dataType) + columnSchema.setColumnName(colName) + val highCardinalityDims = cm.highcardinalitydims.getOrElse(Seq()) + if (highCardinalityDims.contains(colName)) { + encoders.remove(Encoding.DICTIONARY) + } + if (dataType == DataType.TIMESTAMP || dataType == DataType.DATE) { + encoders.add(Encoding.DIRECT_DICTIONARY) + } + columnSchema.setEncodingList(encoders) + val colUniqueIdGenerator = CarbonCommonFactory.getColumnUniqueIdGenerator + val columnUniqueId = colUniqueIdGenerator.generateUniqueId(cm.databaseName, + columnSchema) + columnSchema.setColumnUniqueId(columnUniqueId) + columnSchema.setColumnReferenceId(columnUniqueId) + columnSchema.setColumnar(isCol) + columnSchema.setDimensionColumn(isDimensionCol) + columnSchema.setColumnGroup(colGroup) + columnSchema.setPrecision(precision) + columnSchema.setScale(scale) + columnSchema.setSchemaOrdinal(schemaOrdinal) + columnSchema.setSortColumn(false) + // TODO: Need to fill RowGroupID, converted type + // & Number of Children after DDL finalization + columnSchema + } + + // process create dml fields and create wrapper TableInfo object + def process: TableInfo = { + val LOGGER = LogServiceFactory.getLogService(TableNewProcessor.getClass.getName) + var allColumns = Seq[ColumnSchema]() + var index = 0 + var measureCount = 0 + + // Sort columns should be at the begin of all columns + cm.sortKeyDims.get.foreach { keyDim => + val field = cm.dimCols.find(keyDim equals _.column).get + val encoders = new java.util.ArrayList[Encoding]() + encoders.add(Encoding.DICTIONARY) + val columnSchema: ColumnSchema = getColumnSchema( + DataTypeConverterUtil.convertToCarbonType(field.dataType.getOrElse("")), + field.name.getOrElse(field.column), + index, + isCol = true, + encoders, + isDimensionCol = true, + -1, + field.precision, + field.scale, + field.schemaOrdinal) + columnSchema.setSortColumn(true) + allColumns :+= columnSchema + index = index + 1 + } + + cm.dimCols.foreach(field => { + val sortField = cm.sortKeyDims.get.find(field.column equals _) + if (sortField.isEmpty) { + val encoders = new java.util.ArrayList[Encoding]() + encoders.add(Encoding.DICTIONARY) + val columnSchema: ColumnSchema = getColumnSchema( + DataTypeConverterUtil.convertToCarbonType(field.dataType.getOrElse("")), + field.name.getOrElse(field.column), + index, + isCol = true, + encoders, + isDimensionCol = true, + -1, + field.precision, + field.scale, + field.schemaOrdinal) + allColumns :+= columnSchema + index = index + 1 + if (field.children.isDefined && field.children.get != null) { + columnSchema.setNumberOfChild(field.children.get.size) + allColumns ++= getAllChildren(field.children) + } + } + }) + + cm.msrCols.foreach(field => { + val encoders = new java.util.ArrayList[Encoding]() + val columnSchema: ColumnSchema = getColumnSchema( + DataTypeConverterUtil.convertToCarbonType(field.dataType.getOrElse("")), + field.name.getOrElse(field.column), + index, + isCol = true, + encoders, + isDimensionCol = false, + -1, + field.precision, + field.scale, + field.schemaOrdinal) + allColumns :+= columnSchema + index = index + 1 + measureCount += 1 + }) + + // Check if there is any duplicate measures or dimensions. + // Its based on the dimension name and measure name + allColumns.groupBy(_.getColumnName).foreach(f => if (f._2.size > 1) { + val name = f._1 + LOGGER.error(s"Duplicate column found with name: $name") + LOGGER.audit( + s"Validation failed for Create/Alter Table Operation " + + s"for ${ cm.databaseName }.${ cm.tableName }" + + s"Duplicate column found with name: $name") + sys.error(s"Duplicate dimensions found with name: $name") + }) + + val highCardinalityDims = cm.highcardinalitydims.getOrElse(Seq()) + + checkColGroupsValidity(cm.columnGroups, allColumns, highCardinalityDims) + + updateColumnGroupsInFields(cm.columnGroups, allColumns) + + // Setting the boolean value of useInvertedIndex in column schema + val noInvertedIndexCols = cm.noInvertedIdxCols.getOrElse(Seq()) + LOGGER.info("NoINVERTEDINDEX columns are : " + noInvertedIndexCols.mkString(",")) + for (column <- allColumns) { + // When the column is measure or the specified no inverted index column in DDL, + // set useInvertedIndex to false, otherwise true. + if (noInvertedIndexCols.contains(column.getColumnName) || + cm.msrCols.exists(_.column.equalsIgnoreCase(column.getColumnName))) { + column.setUseInvertedIndex(false) + } else { + column.setUseInvertedIndex(true) + } + } + + // Adding dummy measure if no measure is provided + if (measureCount == 0) { + val encoders = new java.util.ArrayList[Encoding]() + val columnSchema: ColumnSchema = getColumnSchema(DataType.DOUBLE, + CarbonCommonConstants.DEFAULT_INVISIBLE_DUMMY_MEASURE, + index, + true, + encoders, + false, + -1, 0, 0, schemaOrdinal = -1) + columnSchema.setInvisible(true) + allColumns :+= columnSchema + } + val columnValidator = CarbonSparkFactory.getCarbonColumnValidator + columnValidator.validateColumns(allColumns) + + val tableInfo = new TableInfo() + val tableSchema = new TableSchema() + val schemaEvol = new SchemaEvolution() + schemaEvol.setSchemaEvolutionEntryList(new util.ArrayList[SchemaEvolutionEntry]()) + tableSchema.setTableId(UUID.randomUUID().toString) + // populate table properties map + val tablePropertiesMap = new java.util.HashMap[String, String]() + cm.tableProperties.foreach { + x => tablePropertiesMap.put(x._1, x._2) + } + tableSchema.setTableProperties(tablePropertiesMap) + if (cm.bucketFields.isDefined) { + val bucketCols = cm.bucketFields.get.bucketColumns.map { b => + val col = allColumns.find(_.getColumnName.equalsIgnoreCase(b)) + col match { + case Some(colSchema: ColumnSchema) => + if (colSchema.isDimensionColumn && !colSchema.isComplex) { + colSchema + } else { + LOGGER.error(s"Bucket field must be dimension column and " + + s"should not be measure or complex column: ${colSchema.getColumnName}") + sys.error(s"Bucket field must be dimension column and " + + s"should not be measure or complex column: ${colSchema.getColumnName}") + } + case _ => + LOGGER.error(s"Bucket field is not present in table columns") + sys.error(s"Bucket field is not present in table columns") + } + } + tableSchema.setBucketingInfo( + new BucketingInfo(bucketCols.asJava, cm.bucketFields.get.numberOfBuckets)) + } + if (cm.partitionInfo.isDefined) { + val partitionInfo = cm.partitionInfo.get + val PartitionColumnSchema = partitionInfo.getColumnSchemaList.asScala + val partitionCols = allColumns.filter { column => + PartitionColumnSchema.exists(_.getColumnName.equalsIgnoreCase(column.getColumnName)) + }.asJava + partitionInfo.setColumnSchemaList(partitionCols) + tableSchema.setPartitionInfo(partitionInfo) + } + tableSchema.setTableName(cm.tableName) + tableSchema.setListOfColumns(allColumns.asJava) + tableSchema.setSchemaEvalution(schemaEvol) + tableInfo.setDatabaseName(cm.databaseName) + tableInfo.setTableUniqueName(cm.databaseName + "_" + cm.tableName) + tableInfo.setLastUpdatedTime(System.currentTimeMillis()) + tableInfo.setFactTable(tableSchema) + tableInfo + } + + // For checking if the specified col group columns are specified in fields list. + protected def checkColGroupsValidity(colGrps: Seq[String], + allCols: Seq[ColumnSchema], + highCardCols: Seq[String]): Unit = { + if (null != colGrps) { + colGrps.foreach(columngroup => { + val rowCols = columngroup.split(",") + rowCols.foreach(colForGrouping => { + var found: Boolean = false + // check for dimensions + measures + allCols.foreach(eachCol => { + if (eachCol.getColumnName.equalsIgnoreCase(colForGrouping.trim())) { + found = true + } + }) + // check for No Dicitonary dimensions + highCardCols.foreach(noDicCol => { + if (colForGrouping.trim.equalsIgnoreCase(noDicCol)) { + found = true + } + }) + + if (!found) { + sys.error(s"column $colForGrouping is not present in Field list") + } + }) + }) + } + } + + // For updating the col group details for fields. + private def updateColumnGroupsInFields(colGrps: Seq[String], allCols: Seq[ColumnSchema]): Unit = { + if (null != colGrps) { + var colGroupId = -1 + colGrps.foreach(columngroup => { + colGroupId += 1 + val rowCols = columngroup.split(",") + rowCols.foreach(row => { + + allCols.foreach(eachCol => { + + if (eachCol.getColumnName.equalsIgnoreCase(row.trim)) { + eachCol.setColumnGroup(colGroupId) + eachCol.setColumnar(false) + } + }) + }) + }) + } + } +} http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/integration/spark/pom.xml ---------------------------------------------------------------------- diff --git a/integration/spark/pom.xml b/integration/spark/pom.xml index fc13c7a..3f23e1e 100644 --- a/integration/spark/pom.xml +++ b/integration/spark/pom.xml @@ -145,12 +145,6 @@ <version>2.18</version> <!-- Note config is repeated in scalatest config --> <configuration> - <includes> - <include>**/Test*.java</include> - <include>**/*Test.java</include> - <include>**/*TestCase.java</include> - <include>**/*Suite.java</include> - </includes> <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory> <argLine>-Xmx3g -XX:MaxPermSize=512m -XX:ReservedCodeCacheSize=512m</argLine> <systemProperties> @@ -168,7 +162,7 @@ <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory> <junitxml>.</junitxml> <filereports>CarbonTestSuite.txt</filereports> - <argLine>-ea -Xmx3g -XX:MaxPermSize=512m -XX:ReservedCodeCacheSize=512m + <argLine> ${argLine} -ea -Xmx3g -XX:MaxPermSize=512m -XX:ReservedCodeCacheSize=512m </argLine> <stderr /> <environmentVariables> http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/integration/spark2/pom.xml ---------------------------------------------------------------------- diff --git a/integration/spark2/pom.xml b/integration/spark2/pom.xml index 8f17d54..c4572c5 100644 --- a/integration/spark2/pom.xml +++ b/integration/spark2/pom.xml @@ -123,12 +123,6 @@ <version>2.18</version> <!-- Note config is repeated in scalatest config --> <configuration> - <includes> - <include>**/Test*.java</include> - <include>**/*Test.java</include> - <include>**/*TestCase.java</include> - <include>**/*Suite.java</include> - </includes> <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory> <argLine>-Xmx3g -XX:MaxPermSize=512m -XX:ReservedCodeCacheSize=512m</argLine> <systemProperties> @@ -146,7 +140,7 @@ <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory> <junitxml>.</junitxml> <filereports>CarbonTestSuite.txt</filereports> - <argLine>-ea -Xmx3g -XX:MaxPermSize=512m -XX:ReservedCodeCacheSize=512m + <argLine> ${argLine} -ea -Xmx3g -XX:MaxPermSize=512m -XX:ReservedCodeCacheSize=512m </argLine> <stderr /> <environmentVariables> http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/integration/spark2/src/main/scala/org/apache/carbondata/spark/CarbonSparkFactory.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/CarbonSparkFactory.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/CarbonSparkFactory.scala deleted file mode 100644 index 32d121e..0000000 --- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/CarbonSparkFactory.scala +++ /dev/null @@ -1,59 +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.spark - -import org.apache.carbondata.core.metadata.{CarbonTableIdentifier, ColumnIdentifier} -import org.apache.carbondata.core.metadata.schema.table.column.{CarbonDimension, ColumnSchema} - - /** - * Column validator - */ -trait ColumnValidator { - def validateColumns(columns: Seq[ColumnSchema]) -} -/** - * Dictionary related helper service - */ -trait DictionaryDetailService { - def getDictionaryDetail(dictFolderPath: String, primDimensions: Array[CarbonDimension], - table: CarbonTableIdentifier, storePath: String): DictionaryDetail -} - -/** - * Dictionary related detail - */ -case class DictionaryDetail(columnIdentifiers: Array[ColumnIdentifier], - dictFilePaths: Array[String], dictFileExists: Array[Boolean]) - -/** - * Factory class - */ -object CarbonSparkFactory { - /** - * @return column validator - */ - def getCarbonColumnValidator: ColumnValidator = { - new CarbonColumnValidator - } - - /** - * @return dictionary helper - */ - def getDictionaryDetailService: DictionaryDetailService = { - new DictionaryDetailHelper - } -} http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/integration/spark2/src/main/scala/org/apache/carbondata/spark/DictionaryDetailHelper.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/DictionaryDetailHelper.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/DictionaryDetailHelper.scala deleted file mode 100644 index 779ace1..0000000 --- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/DictionaryDetailHelper.scala +++ /dev/null @@ -1,64 +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.spark - -import scala.collection.mutable.HashMap - -import org.apache.carbondata.core.datastore.filesystem.{CarbonFile, CarbonFileFilter} -import org.apache.carbondata.core.datastore.impl.FileFactory -import org.apache.carbondata.core.metadata.{CarbonTableIdentifier, ColumnIdentifier} -import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension -import org.apache.carbondata.core.util.path.{CarbonStorePath, CarbonTablePath} - -class DictionaryDetailHelper extends DictionaryDetailService { - def getDictionaryDetail(dictfolderPath: String, primDimensions: Array[CarbonDimension], - table: CarbonTableIdentifier, storePath: String): DictionaryDetail = { - val carbonTablePath = CarbonStorePath.getCarbonTablePath(storePath, table) - val dictFilePaths = new Array[String](primDimensions.length) - val dictFileExists = new Array[Boolean](primDimensions.length) - val columnIdentifier = new Array[ColumnIdentifier](primDimensions.length) - - val fileType = FileFactory.getFileType(dictfolderPath) - // Metadata folder - val metadataDirectory = FileFactory.getCarbonFile(dictfolderPath, fileType) - // need list all dictionary file paths with exists flag - val fileNamesMap = new HashMap[String, Int] - if (metadataDirectory.exists()) { - val carbonFiles = metadataDirectory.listFiles(new CarbonFileFilter { - @Override def accept(pathname: CarbonFile): Boolean = { - CarbonTablePath.isDictionaryFile(pathname) - } - }) - // 2 put dictionary file names to fileNamesMap - for (i <- 0 until carbonFiles.length) { - fileNamesMap.put(carbonFiles(i).getName, i) - } - } - // 3 lookup fileNamesMap, if file name is in fileNamesMap, file is exists, or not. - primDimensions.zipWithIndex.foreach { f => - columnIdentifier(f._2) = f._1.getColumnIdentifier - dictFilePaths(f._2) = carbonTablePath.getDictionaryFilePath(f._1.getColumnId) - dictFileExists(f._2) = - fileNamesMap.get(CarbonTablePath.getDictionaryFileName(f._1.getColumnId)) match { - case None => false - case Some(_) => true - } - } - - DictionaryDetail(columnIdentifier, dictFilePaths, dictFileExists) - } -} http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala index 303a8b6..4f4515d 100644 --- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala +++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala @@ -17,8 +17,6 @@ package org.apache.spark.sql -import java.io.{ByteArrayOutputStream, DataOutputStream} - import scala.collection.mutable.ArrayBuffer import org.apache.spark.CarbonInputMetrics @@ -26,6 +24,7 @@ import org.apache.spark.rdd.RDD import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.execution.command.LoadTableByInsert import org.apache.spark.sql.hive.CarbonRelation +import org.apache.spark.sql.optimizer.CarbonFilters import org.apache.spark.sql.sources.{BaseRelation, Filter, InsertableRelation} import org.apache.spark.sql.types.StructType @@ -35,8 +34,7 @@ import org.apache.carbondata.core.metadata.schema.table.CarbonTable import org.apache.carbondata.core.scan.expression.Expression import org.apache.carbondata.core.scan.expression.logical.AndExpression import org.apache.carbondata.core.util.{CarbonSessionInfo, ThreadLocalSessionInfo} -import org.apache.carbondata.hadoop.{CarbonProjection, InputMetricsStats} -import org.apache.carbondata.spark.CarbonFilters +import org.apache.carbondata.hadoop.CarbonProjection import org.apache.carbondata.spark.rdd.CarbonScanRDD case class CarbonDatasourceHadoopRelation( http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonScan.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonScan.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonScan.scala index b91acef..0806421 100644 --- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonScan.scala +++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonScan.scala @@ -22,9 +22,9 @@ import scala.collection.mutable.ArrayBuffer import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.hive.CarbonRelation +import org.apache.spark.sql.optimizer.CarbonFilters import org.apache.carbondata.core.scan.model._ -import org.apache.carbondata.spark.CarbonFilters case class CarbonScan( var attributesRaw: Seq[Attribute], http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/integration/spark2/src/main/scala/org/apache/spark/sql/SparkUnknownExpression.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/SparkUnknownExpression.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/SparkUnknownExpression.scala index c472777..50d7dba 100644 --- a/integration/spark2/src/main/scala/org/apache/spark/sql/SparkUnknownExpression.scala +++ b/integration/spark2/src/main/scala/org/apache/spark/sql/SparkUnknownExpression.scala @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.carbondata.spark +package org.apache.spark.sql import java.util.{ArrayList, List} http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonHiveCommands.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonHiveCommands.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonHiveCommands.scala deleted file mode 100644 index 2731104..0000000 --- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonHiveCommands.scala +++ /dev/null @@ -1,78 +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.spark.sql.hive.execution.command - -import org.apache.spark.sql.{CarbonEnv, Row, SparkSession} -import org.apache.spark.sql.catalyst.TableIdentifier -import org.apache.spark.sql.execution.command.{CarbonDropTableCommand, DropDatabaseCommand, ResetCommand, RunnableCommand, SetCommand} - -import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil} - -case class CarbonDropDatabaseCommand(command: DropDatabaseCommand) - extends RunnableCommand { - - override val output = command.output - - override def run(sparkSession: SparkSession): Seq[Row] = { - val dbName = command.databaseName - var tablesInDB: Seq[TableIdentifier] = null - if (sparkSession.sessionState.catalog.listDatabases().exists(_.equalsIgnoreCase(dbName))) { - tablesInDB = sparkSession.sessionState.catalog.listTables(dbName) - } - // DropHiveDB command will fail if cascade is false and one or more table exists in database - val rows = command.run(sparkSession) - if (command.cascade && tablesInDB != null) { - tablesInDB.foreach { tableName => - CarbonDropTableCommand(true, tableName.database, tableName.table).run(sparkSession) - } - } - CarbonUtil.dropDatabaseDirectory(dbName.toLowerCase, - CarbonEnv.getInstance(sparkSession).storePath) - rows - } -} - -case class CarbonSetCommand(command: SetCommand) - extends RunnableCommand { - - override val output = command.output - - override def run(sparkSession: SparkSession): Seq[Row] = { - val sessionParms = CarbonEnv.getInstance(sparkSession).carbonSessionInfo.getSessionParams - command.kv match { - case Some((key, Some(value))) => - val isCarbonProperty: Boolean = CarbonProperties.getInstance().isCarbonProperty(key) - if (isCarbonProperty) { - sessionParms.addProperty(key, value) - } - case _ => - - } - command.run(sparkSession) - } -} - -case class CarbonResetCommand() - extends RunnableCommand { - override val output = ResetCommand.output - - override def run(sparkSession: SparkSession): Seq[Row] = { - CarbonEnv.getInstance(sparkSession).carbonSessionInfo.getSessionParams.clear() - ResetCommand.run(sparkSession) - } -} http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/integration/spark2/src/main/scala/org/apache/spark/sql/hive/execution/command/CarbonHiveCommands.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/execution/command/CarbonHiveCommands.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/execution/command/CarbonHiveCommands.scala new file mode 100644 index 0000000..f435fa6 --- /dev/null +++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/execution/command/CarbonHiveCommands.scala @@ -0,0 +1,78 @@ +/* + * 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.spark.sql.hive.execution.command + +import org.apache.spark.sql.{CarbonEnv, Row, SparkSession} +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.execution.command._ + +import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil} + +case class CarbonDropDatabaseCommand(command: DropDatabaseCommand) + extends RunnableCommand { + + override val output = command.output + + override def run(sparkSession: SparkSession): Seq[Row] = { + val dbName = command.databaseName + var tablesInDB: Seq[TableIdentifier] = null + if (sparkSession.sessionState.catalog.listDatabases().exists(_.equalsIgnoreCase(dbName))) { + tablesInDB = sparkSession.sessionState.catalog.listTables(dbName) + } + // DropHiveDB command will fail if cascade is false and one or more table exists in database + val rows = command.run(sparkSession) + if (command.cascade && tablesInDB != null) { + tablesInDB.foreach { tableName => + CarbonDropTableCommand(true, tableName.database, tableName.table).run(sparkSession) + } + } + CarbonUtil.dropDatabaseDirectory(dbName.toLowerCase, + CarbonEnv.getInstance(sparkSession).storePath) + rows + } +} + +case class CarbonSetCommand(command: SetCommand) + extends RunnableCommand { + + override val output = command.output + + override def run(sparkSession: SparkSession): Seq[Row] = { + val sessionParms = CarbonEnv.getInstance(sparkSession).carbonSessionInfo.getSessionParams + command.kv match { + case Some((key, Some(value))) => + val isCarbonProperty: Boolean = CarbonProperties.getInstance().isCarbonProperty(key) + if (isCarbonProperty) { + sessionParms.addProperty(key, value) + } + case _ => + + } + command.run(sparkSession) + } +} + +case class CarbonResetCommand() + extends RunnableCommand { + override val output = ResetCommand.output + + override def run(sparkSession: SparkSession): Seq[Row] = { + CarbonEnv.getInstance(sparkSession).carbonSessionInfo.getSessionParams.clear() + ResetCommand.run(sparkSession) + } +} http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala index 31a3208..73c9760 100644 --- a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala +++ b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala @@ -15,13 +15,13 @@ * limitations under the License. */ -package org.apache.carbondata.spark +package org.apache.spark.sql.optimizer import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.execution.CastExpressionOptimization -import org.apache.spark.sql.optimizer.AttributeReferenceWrapper import org.apache.spark.sql.CarbonBoundReference import org.apache.spark.sql.CastExpr +import org.apache.spark.sql.SparkUnknownExpression import org.apache.spark.sql.sources import org.apache.spark.sql.types._ import org.apache.spark.sql.CarbonContainsWith @@ -33,6 +33,7 @@ import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn import org.apache.carbondata.core.scan.expression.{ColumnExpression => CarbonColumnExpression, Expression => CarbonExpression, LiteralExpression => CarbonLiteralExpression} import org.apache.carbondata.core.scan.expression.conditional._ import org.apache.carbondata.core.scan.expression.logical.{AndExpression, FalseExpression, OrExpression} +import org.apache.carbondata.spark.CarbonAliasDecoderRelation import org.apache.carbondata.spark.util.CarbonScalaUtil http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala index 961874e..0dca0d4 100644 --- a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala +++ b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala @@ -35,7 +35,7 @@ import org.apache.carbondata.common.logging.LogServiceFactory import org.apache.carbondata.core.constants.CarbonCommonConstants import org.apache.carbondata.core.stats.QueryStatistic import org.apache.carbondata.core.util.CarbonTimeStatisticsFactory -import org.apache.carbondata.spark.{CarbonAliasDecoderRelation, CarbonFilters} +import org.apache.carbondata.spark.CarbonAliasDecoderRelation /** http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/pom.xml ---------------------------------------------------------------------- diff --git a/pom.xml b/pom.xml index df3c5bd..63f8b2f 100644 --- a/pom.xml +++ b/pom.xml @@ -328,6 +328,67 @@ <outputEncoding>${project.build.sourceEncoding}</outputEncoding> </configuration> </plugin> + <plugin> + <groupId>org.jacoco</groupId> + <artifactId>jacoco-maven-plugin</artifactId> + <version>0.7.9</version> + <executions> + <execution> + <id>default-prepare-agent</id> + <goals> + <goal>prepare-agent</goal> + </goals> + </execution> + <execution> + <id>default-prepare-agent-integration</id> + <goals> + <goal>prepare-agent-integration</goal> + </goals> + </execution> + <execution> + <id>default-report</id> + <goals> + <goal>report</goal> + </goals> + </execution> + <execution> + <id>default-report-integration</id> + <goals> + <goal>report-integration</goal> + </goals> + </execution> + <execution> + <id>default-check</id> + <goals> + <goal>check</goal> + </goals> + <configuration> + <append>true</append> + <excludes> + <exclude>**/*SparkUnknownExpression*.class</exclude> + </excludes> + <includes> + <include>**/org.apache.*</include> + </includes> + <rules> + <!-- implementation is needed only for Maven 2 --> + <rule implementation="org.jacoco.maven.RuleConfiguration"> + <element>BUNDLE</element> + <limits> + <!-- implementation is needed only for Maven 2 --> + <limit implementation="org.jacoco.report.check.Limit"> + <counter>COMPLEXITY</counter> + <value>COVEREDRATIO</value> + <minimum>0.10</minimum> + </limit> + </limits> + </rule> + </rules> + </configuration> + </execution> + </executions> + </plugin> + </plugins> </build> @@ -397,6 +458,39 @@ <module>examples/spark</module> <module>examples/flink</module> </modules> + <build> + <plugins> + <plugin> + <groupId>org.eluder.coveralls</groupId> + <artifactId>coveralls-maven-plugin</artifactId> + <version>4.3.0</version> + <configuration> + <repoToken>Q8mcBW8dw34wjNiPAMx0SFQidspIwHw8g</repoToken> + <sourceEncoding>UTF-8</sourceEncoding> + <jacocoReports> + <jacocoReport>${basedir}/target/carbondata-coverage-report/carbondata-coverage-report.xml + </jacocoReport> + </jacocoReports> + <sourceDirectories> + <sourceDirectory>${basedir}/common/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/core/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/processing/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/hadoop/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/integration/spark-common/src/main/scala</sourceDirectory> + <sourceDirectory>${basedir}/integration/spark-common/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/integration/spark-common-test/src/main/scala</sourceDirectory> + <sourceDirectory>${basedir}/integration/spark-common-test/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/integration/hive/src/main/scala</sourceDirectory> + <sourceDirectory>${basedir}/integration/hive/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/integration/presto/src/main/scala</sourceDirectory> + <sourceDirectory>${basedir}/integration/presto/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/integration/spark/src/main/scala</sourceDirectory> + <sourceDirectory>${basedir}/integration/spark/src/main/java</sourceDirectory> + </sourceDirectories> + </configuration> + </plugin> + </plugins> + </build> </profile> <profile> <id>spark-2.1</id> @@ -414,6 +508,39 @@ <module>integration/presto</module> <module>examples/spark2</module> </modules> + <build> + <plugins> + <plugin> + <groupId>org.eluder.coveralls</groupId> + <artifactId>coveralls-maven-plugin</artifactId> + <version>4.3.0</version> + <configuration> + <repoToken>Q8mcBW8dw34wjNiPAMx0SFQidspIwHw8g</repoToken> + <sourceEncoding>UTF-8</sourceEncoding> + <jacocoReports> + <jacocoReport>${basedir}/target/carbondata-coverage-report/carbondata-coverage-report.xml + </jacocoReport> + </jacocoReports> + <sourceDirectories> + <sourceDirectory>${basedir}/common/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/core/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/processing/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/hadoop/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/integration/spark2/src/main/scala</sourceDirectory> + <sourceDirectory>${basedir}/integration/spark2/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/integration/spark-common/src/main/scala</sourceDirectory> + <sourceDirectory>${basedir}/integration/spark-common/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/integration/spark-common-test/src/main/scala</sourceDirectory> + <sourceDirectory>${basedir}/integration/spark-common-test/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/integration/hive/src/main/scala</sourceDirectory> + <sourceDirectory>${basedir}/integration/hive/src/main/java</sourceDirectory> + <sourceDirectory>${basedir}/integration/presto/src/main/scala</sourceDirectory> + <sourceDirectory>${basedir}/integration/presto/src/main/java</sourceDirectory> + </sourceDirectories> + </configuration> + </plugin> + </plugins> + </build> </profile> <profile> <id>findbugs</id> http://git-wip-us.apache.org/repos/asf/carbondata/blob/ab9a4fc7/processing/src/test/java/org/apache/carbondata/lcm/locks/ZooKeeperLockingTest.java ---------------------------------------------------------------------- diff --git a/processing/src/test/java/org/apache/carbondata/lcm/locks/ZooKeeperLockingTest.java b/processing/src/test/java/org/apache/carbondata/lcm/locks/ZooKeeperLockingTest.java index 6ee6287..29293df 100644 --- a/processing/src/test/java/org/apache/carbondata/lcm/locks/ZooKeeperLockingTest.java +++ b/processing/src/test/java/org/apache/carbondata/lcm/locks/ZooKeeperLockingTest.java @@ -82,15 +82,15 @@ public class ZooKeeperLockingTest { throws IllegalArgumentException, IllegalAccessException, NoSuchFieldException, SecurityException { - final CarbonProperties cp = CarbonProperties.getInstance(); - new NonStrictExpectations(cp) { - { - cp.getProperty("/CarbonLocks"); - result = "/carbontests"; - cp.getProperty("spark.deploy.zookeeper.url"); - result = "127.0.0.1:" + freePort; - } - }; +// final CarbonProperties cp = CarbonProperties.getInstance(); +// new NonStrictExpectations(cp) { +// { +// cp.getProperty("/CarbonLocks"); +// result = "/carbontests"; +// cp.getProperty("spark.deploy.zookeeper.url"); +// result = "127.0.0.1:" + freePort; +// } +// }; ZookeeperInit zki = ZookeeperInit.getInstance("127.0.0.1:" + freePort);
