3669

Build # 3669

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/e341065683a70017d7a5eedbe2df0f039080662c">e34106568<a href="https://github.com/apache/carbondata/commit/e341065683a70017d7a5eedbe2df0f039080662c">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/c9958d5d3244a99e0a3c531738f5f19988ca8cdb">c9958d5d3</a>">c9958d5d3</a><a href="https://github.com/apache/carbondata/commit/e341065683a70017d7a5eedbe2df0f039080662c"> into 9d0209226">9d0209226</a>

Run Details

11 of 11 new or added lines in 2 files covered. (100.0%)

63985 of 82715 relevant lines covered (77.36%)

1.01 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.09

/integration/spark2/src/main/scala/org/apache/spark/util/DataMapUtil.scala

/*
 * 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.util

import java.io.IOException
import java.util

import scala.collection.JavaConverters._
import scala.collection.mutable

import org.apache.spark.sql.execution.command.{DataMapField, Field}

import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.datastore.compression.CompressorFactory
import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider.MV
import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}

/**
 * Utility class for keeping all the utility methods common for pre-aggregate and mv datamap
 */
object DataMapUtil {

  def inheritTablePropertiesFromMainTable(parentTable: CarbonTable,
      fields: Seq[Field],
      fieldRelationMap: scala.collection.mutable.LinkedHashMap[Field, DataMapField],
      tableProperties: mutable.Map[String, String]): Unit = {
    var neworder = Seq[String]()
    val parentOrder = parentTable.getSortColumns(parentTable.getTableName).asScala
    parentOrder.foreach(parentcol =>
      fields.filter(col => fieldRelationMap(col).aggregateFunction.isEmpty &&
                           fieldRelationMap(col).columnTableRelationList.size == 1 &&
                           parentcol.equals(fieldRelationMap(col).
                             columnTableRelationList.get(0).parentColumnName))
        .map(cols => neworder :+= cols.column))
    tableProperties.put(CarbonCommonConstants.SORT_COLUMNS, neworder.mkString(","))
    val sort_scope = parentTable.getTableInfo.getFactTable.getTableProperties.asScala
      .get("sort_scope")
    if (sort_scope.isDefined) {
      tableProperties.put("sort_scope", sort_scope.get)
    }
    tableProperties
      .put(CarbonCommonConstants.TABLE_BLOCKSIZE, parentTable.getBlockSizeInMB.toString)
    tableProperties.put(CarbonCommonConstants.FLAT_FOLDER,
      parentTable.getTableInfo.getFactTable.getTableProperties.asScala.getOrElse(
        CarbonCommonConstants.FLAT_FOLDER, CarbonCommonConstants.DEFAULT_FLAT_FOLDER))

    // Datamap table name and columns are automatically added prefix with parent table name
    // in carbon. For convenient, users can type column names same as the ones in select statement
    // when config dmproperties, and here we update column names with prefix.
    // If longStringColumn is not present in dm properties then we take long_string_columns from
    // the parent table.
    var longStringColumn = tableProperties.get(CarbonCommonConstants.LONG_STRING_COLUMNS)
    if (longStringColumn.isEmpty) {
      val longStringColumnInParents = parentTable.getTableInfo.getFactTable.getTableProperties
        .asScala
        .getOrElse(CarbonCommonConstants.LONG_STRING_COLUMNS, "").split(",").map(_.trim)
      val varcharDatamapFields = scala.collection.mutable.ArrayBuffer.empty[String]
      fieldRelationMap foreach (fields => {
        val aggFunc = fields._2.aggregateFunction
        val relationList = fields._2.columnTableRelationList
        // check if columns present in datamap are long_string_col in parent table. If they are
        // long_string_columns in parent, make them long_string_columns in datamap
        if (aggFunc.isEmpty && relationList.size == 1 && longStringColumnInParents
          .contains(relationList.head.head.parentColumnName)) {
          varcharDatamapFields += relationList.head.head.parentColumnName
        }
      })
      if (!varcharDatamapFields.isEmpty) {
        longStringColumn = Option(varcharDatamapFields.mkString(","))
      }
    }

    if (longStringColumn != None) {
      val fieldNames = fields.map(_.column)
      val newLongStringColumn = longStringColumn.get.split(",").map(_.trim).map { colName =>
        val newColName = parentTable.getTableName.toLowerCase() + "_" + colName
        if (!fieldNames.contains(newColName)) {
          throw new MalformedDataMapCommandException(
            CarbonCommonConstants.LONG_STRING_COLUMNS.toUpperCase() + ":" + colName
            + " does not in datamap")
        }
        newColName
      }
      tableProperties.put(CarbonCommonConstants.LONG_STRING_COLUMNS,
        newLongStringColumn.mkString(","))
    }
    // inherit compressor property
    tableProperties
      .put(CarbonCommonConstants.COMPRESSOR,
        parentTable.getTableInfo.getFactTable.getTableProperties
          .getOrDefault(CarbonCommonConstants.COMPRESSOR,
            CompressorFactory.getInstance().getCompressor.getName))

    // inherit the local dictionary properties of main parent table
    tableProperties
      .put(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE,
        parentTable.getTableInfo.getFactTable.getTableProperties.asScala
          .getOrElse(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE, "false"))
    tableProperties
      .put(CarbonCommonConstants.LOCAL_DICTIONARY_THRESHOLD,
        parentTable.getTableInfo.getFactTable.getTableProperties.asScala
          .getOrElse(CarbonCommonConstants.LOCAL_DICTIONARY_THRESHOLD,
            CarbonCommonConstants.LOCAL_DICTIONARY_THRESHOLD_DEFAULT))
    val parentDictInclude = parentTable.getTableInfo.getFactTable.getTableProperties.asScala
      .getOrElse(CarbonCommonConstants.LOCAL_DICTIONARY_INCLUDE, "").split(",")

    val parentDictExclude = parentTable.getTableInfo.getFactTable.getTableProperties.asScala
      .getOrElse(CarbonCommonConstants.LOCAL_DICTIONARY_EXCLUDE, "").split(",")

    val parentGlobalDictInclude = parentTable.getTableInfo.getFactTable.getTableProperties.asScala
      .getOrElse(CarbonCommonConstants.DICTIONARY_INCLUDE, "").split(",")

    val parentGlobalDictExclude = parentTable.getTableInfo.getFactTable.getTableProperties.asScala
      .getOrElse(CarbonCommonConstants.DICTIONARY_EXCLUDE, "").split(",")

    val newLocalDictInclude = getDataMapColumns(parentDictInclude, fields, fieldRelationMap)

    val newLocalDictExclude = getDataMapColumns(parentDictExclude, fields, fieldRelationMap)

    val newGlobalDictInclude = getDataMapColumns(parentGlobalDictInclude, fields, fieldRelationMap)

    val newGlobalDictExclude = getDataMapColumns(parentGlobalDictExclude, fields, fieldRelationMap)

    if (newLocalDictInclude.nonEmpty) {
      tableProperties
        .put(CarbonCommonConstants.LOCAL_DICTIONARY_INCLUDE, newLocalDictInclude.mkString(","))
    }
    if (newLocalDictExclude.nonEmpty) {
      tableProperties
        .put(CarbonCommonConstants.LOCAL_DICTIONARY_EXCLUDE, newLocalDictExclude.mkString(","))
    }

    if (newGlobalDictInclude.nonEmpty) {
      tableProperties
        .put(CarbonCommonConstants.DICTIONARY_INCLUDE, newGlobalDictInclude.mkString(","))
    }
    if (newGlobalDictExclude.nonEmpty) {
      tableProperties
        .put(CarbonCommonConstants.DICTIONARY_EXCLUDE, newGlobalDictExclude.mkString(","))
    }

    val parentInvertedIndex = parentTable.getTableInfo.getFactTable.getTableProperties.asScala
      .getOrElse(CarbonCommonConstants.INVERTED_INDEX, "").split(",")

    val newInvertedIndex = getDataMapColumns(parentInvertedIndex, fields, fieldRelationMap)

    val parentNoInvertedIndex = parentTable.getTableInfo.getFactTable.getTableProperties.asScala
      .getOrElse(CarbonCommonConstants.NO_INVERTED_INDEX, "").split(",")

    val newNoInvertedIndex = getDataMapColumns(parentNoInvertedIndex, fields, fieldRelationMap)

    if (newInvertedIndex.nonEmpty) {
      tableProperties
        .put(CarbonCommonConstants.INVERTED_INDEX, newInvertedIndex.mkString(","))
    }
    if (newNoInvertedIndex.nonEmpty) {
      tableProperties
        .put(CarbonCommonConstants.NO_INVERTED_INDEX, newNoInvertedIndex.mkString(","))
    }

  }

  private def getDataMapColumns(parentColumns: Array[String], fields: Seq[Field],
      fieldRelationMap: scala.collection.mutable.LinkedHashMap[Field, DataMapField]) = {
    val dataMapColumns = parentColumns.flatMap(parentcol =>
      fields.collect {
        case col if fieldRelationMap(col).aggregateFunction.isEmpty &&
                    fieldRelationMap(col).columnTableRelationList.size == 1 &&
                    parentcol.equals(fieldRelationMap(col).
                      columnTableRelationList.get.head.parentColumnName) =>
          col.column
      })
    dataMapColumns
  }

  /**
   * Return true if MV datamap present in the specified table
   */
  @throws[IOException]
  def hasMVDataMap(carbonTable: CarbonTable): Boolean = {
    val dataMapSchemaList = DataMapStoreManager.getInstance.
      getDataMapSchemasOfTable(carbonTable).asScala
    dataMapSchemaList.foreach { dataMapSchema =>
      if (dataMapSchema.getProviderName.equalsIgnoreCase(MV.toString)) {
        return true
      }
    }
    false
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.spark.util
19
20	import java.io.IOException
21	import java.util
22
23	import scala.collection.JavaConverters._
24	import scala.collection.mutable
25
26	import org.apache.spark.sql.execution.command.{DataMapField, Field}
27
28	import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException
29	import org.apache.carbondata.core.constants.CarbonCommonConstants
30	import org.apache.carbondata.core.datamap.DataMapStoreManager
31	import org.apache.carbondata.core.datastore.compression.CompressorFactory
32	import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider.MV
33	import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}
34
35	/**
36	* Utility class for keeping all the utility methods common for pre-aggregate and mv datamap
37	*/
38	object DataMapUtil {
39
40	def inheritTablePropertiesFromMainTable(parentTable: CarbonTable,
41	fields: Seq[Field],
42	fieldRelationMap: scala.collection.mutable.LinkedHashMap[Field, DataMapField],
43	tableProperties: mutable.Map[String, String]): Unit = {	2✔
44	var neworder = Seq[String]()	2✔
45	val parentOrder = parentTable.getSortColumns(parentTable.getTableName).asScala	2✔
46	parentOrder.foreach(parentcol =>	2✔
47	fields.filter(col => fieldRelationMap(col).aggregateFunction.isEmpty &&	1✔
48	fieldRelationMap(col).columnTableRelationList.size == 1 &&	1✔
49	parentcol.equals(fieldRelationMap(col).	1✔
50	columnTableRelationList.get(0).parentColumnName))	1✔
51	.map(cols => neworder :+= cols.column))	1✔
52	tableProperties.put(CarbonCommonConstants.SORT_COLUMNS, neworder.mkString(","))	2✔
53	val sort_scope = parentTable.getTableInfo.getFactTable.getTableProperties.asScala	2✔
54	.get("sort_scope")	2✔
55	if (sort_scope.isDefined) {	2✔
56	tableProperties.put("sort_scope", sort_scope.get)	1✔
57	}
58	tableProperties	2✔
59	.put(CarbonCommonConstants.TABLE_BLOCKSIZE, parentTable.getBlockSizeInMB.toString)	2✔
60	tableProperties.put(CarbonCommonConstants.FLAT_FOLDER,	2✔
61	parentTable.getTableInfo.getFactTable.getTableProperties.asScala.getOrElse(	2✔
62	CarbonCommonConstants.FLAT_FOLDER, CarbonCommonConstants.DEFAULT_FLAT_FOLDER))	2✔
63
64	// Datamap table name and columns are automatically added prefix with parent table name
65	// in carbon. For convenient, users can type column names same as the ones in select statement
66	// when config dmproperties, and here we update column names with prefix.
67	// If longStringColumn is not present in dm properties then we take long_string_columns from
68	// the parent table.
69	var longStringColumn = tableProperties.get(CarbonCommonConstants.LONG_STRING_COLUMNS)	2✔
70	if (longStringColumn.isEmpty) {	2✔
71	val longStringColumnInParents = parentTable.getTableInfo.getFactTable.getTableProperties	2✔
72	.asScala	2✔
73	.getOrElse(CarbonCommonConstants.LONG_STRING_COLUMNS, "").split(",").map(_.trim)	2✔
74	val varcharDatamapFields = scala.collection.mutable.ArrayBuffer.empty[String]	2✔
75	fieldRelationMap foreach (fields => {	2✔
76	val aggFunc = fields._2.aggregateFunction	2✔
77	val relationList = fields._2.columnTableRelationList	2✔
78	// check if columns present in datamap are long_string_col in parent table. If they are
79	// long_string_columns in parent, make them long_string_columns in datamap
80	if (aggFunc.isEmpty && relationList.size == 1 && longStringColumnInParents	2✔
81	.contains(relationList.head.head.parentColumnName)) {	2✔
82	varcharDatamapFields += relationList.head.head.parentColumnName	1✔
83	}
84	})
85	if (!varcharDatamapFields.isEmpty) {	2✔
86	longStringColumn = Option(varcharDatamapFields.mkString(","))	1✔
87	}
88	}
89
90	if (longStringColumn != None) {	2✔
91	val fieldNames = fields.map(_.column)	1✔
92	val newLongStringColumn = longStringColumn.get.split(",").map(_.trim).map { colName =>	1✔
93	val newColName = parentTable.getTableName.toLowerCase() + "_" + colName	1✔
94	if (!fieldNames.contains(newColName)) {	1✔
95	throw new MalformedDataMapCommandException(	×
96	CarbonCommonConstants.LONG_STRING_COLUMNS.toUpperCase() + ":" + colName	×
97	+ " does not in datamap")	×
98	}
99	newColName	1✔
100	}
101	tableProperties.put(CarbonCommonConstants.LONG_STRING_COLUMNS,	1✔
102	newLongStringColumn.mkString(","))	1✔
103	}
104	// inherit compressor property
105	tableProperties	2✔
106	.put(CarbonCommonConstants.COMPRESSOR,	2✔
107	parentTable.getTableInfo.getFactTable.getTableProperties	2✔
108	.getOrDefault(CarbonCommonConstants.COMPRESSOR,	2✔
109	CompressorFactory.getInstance().getCompressor.getName))	2✔
110
111	// inherit the local dictionary properties of main parent table
112	tableProperties	2✔
113	.put(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE,	2✔
114	parentTable.getTableInfo.getFactTable.getTableProperties.asScala	2✔
115	.getOrElse(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE, "false"))	2✔
116	tableProperties	2✔
117	.put(CarbonCommonConstants.LOCAL_DICTIONARY_THRESHOLD,	2✔
118	parentTable.getTableInfo.getFactTable.getTableProperties.asScala	2✔
119	.getOrElse(CarbonCommonConstants.LOCAL_DICTIONARY_THRESHOLD,	2✔
120	CarbonCommonConstants.LOCAL_DICTIONARY_THRESHOLD_DEFAULT))	2✔
121	val parentDictInclude = parentTable.getTableInfo.getFactTable.getTableProperties.asScala	2✔
122	.getOrElse(CarbonCommonConstants.LOCAL_DICTIONARY_INCLUDE, "").split(",")	2✔
123
124	val parentDictExclude = parentTable.getTableInfo.getFactTable.getTableProperties.asScala	2✔
125	.getOrElse(CarbonCommonConstants.LOCAL_DICTIONARY_EXCLUDE, "").split(",")	2✔
126
127	val parentGlobalDictInclude = parentTable.getTableInfo.getFactTable.getTableProperties.asScala	2✔
128	.getOrElse(CarbonCommonConstants.DICTIONARY_INCLUDE, "").split(",")	2✔
129
130	val parentGlobalDictExclude = parentTable.getTableInfo.getFactTable.getTableProperties.asScala	2✔
131	.getOrElse(CarbonCommonConstants.DICTIONARY_EXCLUDE, "").split(",")	2✔
132
133	val newLocalDictInclude = getDataMapColumns(parentDictInclude, fields, fieldRelationMap)	2✔
134
135	val newLocalDictExclude = getDataMapColumns(parentDictExclude, fields, fieldRelationMap)	2✔
136
137	val newGlobalDictInclude = getDataMapColumns(parentGlobalDictInclude, fields, fieldRelationMap)	2✔
138
139	val newGlobalDictExclude = getDataMapColumns(parentGlobalDictExclude, fields, fieldRelationMap)	2✔
140
141	if (newLocalDictInclude.nonEmpty) {	2✔
142	tableProperties	×
143	.put(CarbonCommonConstants.LOCAL_DICTIONARY_INCLUDE, newLocalDictInclude.mkString(","))	×
144	}
145	if (newLocalDictExclude.nonEmpty) {	2✔
146	tableProperties	×
147	.put(CarbonCommonConstants.LOCAL_DICTIONARY_EXCLUDE, newLocalDictExclude.mkString(","))	×
148	}
149
150	if (newGlobalDictInclude.nonEmpty) {	2✔
151	tableProperties	1✔
152	.put(CarbonCommonConstants.DICTIONARY_INCLUDE, newGlobalDictInclude.mkString(","))	1✔
153	}
154	if (newGlobalDictExclude.nonEmpty) {	2✔
155	tableProperties	1✔
156	.put(CarbonCommonConstants.DICTIONARY_EXCLUDE, newGlobalDictExclude.mkString(","))	1✔
157	}
158
159	val parentInvertedIndex = parentTable.getTableInfo.getFactTable.getTableProperties.asScala	2✔
160	.getOrElse(CarbonCommonConstants.INVERTED_INDEX, "").split(",")	2✔
161
162	val newInvertedIndex = getDataMapColumns(parentInvertedIndex, fields, fieldRelationMap)	2✔
163
164	val parentNoInvertedIndex = parentTable.getTableInfo.getFactTable.getTableProperties.asScala	2✔
165	.getOrElse(CarbonCommonConstants.NO_INVERTED_INDEX, "").split(",")	2✔
166
167	val newNoInvertedIndex = getDataMapColumns(parentNoInvertedIndex, fields, fieldRelationMap)	2✔
168
169	if (newInvertedIndex.nonEmpty) {	2✔
170	tableProperties	×
171	.put(CarbonCommonConstants.INVERTED_INDEX, newInvertedIndex.mkString(","))	×
172	}
173	if (newNoInvertedIndex.nonEmpty) {	2✔
174	tableProperties	×
175	.put(CarbonCommonConstants.NO_INVERTED_INDEX, newNoInvertedIndex.mkString(","))	×
176	}
177
178	}
179
180	private def getDataMapColumns(parentColumns: Array[String], fields: Seq[Field],
181	fieldRelationMap: scala.collection.mutable.LinkedHashMap[Field, DataMapField]) = {
182	val dataMapColumns = parentColumns.flatMap(parentcol =>	2✔
183	fields.collect {	2✔
184	case col if fieldRelationMap(col).aggregateFunction.isEmpty &&	2✔
185	fieldRelationMap(col).columnTableRelationList.size == 1 &&	2✔
186	parentcol.equals(fieldRelationMap(col).	2✔
187	columnTableRelationList.get.head.parentColumnName) =>	2✔
188	col.column	1✔
189	})
190	dataMapColumns	2✔
191	}
192
193	/**
194	* Return true if MV datamap present in the specified table
195	*/
196	@throws[IOException]
197	def hasMVDataMap(carbonTable: CarbonTable): Boolean = {	2✔
198	val dataMapSchemaList = DataMapStoreManager.getInstance.	2✔
199	getDataMapSchemasOfTable(carbonTable).asScala	2✔
200	dataMapSchemaList.foreach { dataMapSchema =>	2✔
201	if (dataMapSchema.getProviderName.equalsIgnoreCase(MV.toString)) {	2✔
202	return true	×
203	}
204	}
205	false	2✔
206	}
207	}	2✔

apache / carbondata / 3669

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous