Repository: carbondata Updated Branches: refs/heads/master dc2931917 -> 36e14e515
[CARBONDATA-2587][CARBONDATA-2588]added test cases for local dictioanry load support test cases are added for local dictionary load support. All the scenarios are covered like enable, disable, valid and invalid threshold, fallback scenarios and complexTypes This closes #2424 Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/36e14e51 Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/36e14e51 Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/36e14e51 Branch: refs/heads/master Commit: 36e14e5151049a4d452ab2e591089f1e42bf4650 Parents: dc29319 Author: akashrn5 <[email protected]> Authored: Wed Jun 27 17:18:47 2018 +0530 Committer: kunal642 <[email protected]> Committed: Wed Jul 11 13:50:00 2018 +0530 ---------------------------------------------------------------------- .../blocklet/BlockletEncodedColumnPage.java | 2 +- .../datastore/page/LocalDictColumnPage.java | 4 +- .../ColumnLocalDictionaryGenerator.java | 2 +- .../TestPageLevelDictionary.java | 16 + .../src/test/resources/localdictionary.csv | 8 + .../LocalDictionarySupportLoadTableTest.scala | 309 +++++++++++++++++++ 6 files changed, 337 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/carbondata/blob/36e14e51/core/src/main/java/org/apache/carbondata/core/datastore/blocklet/BlockletEncodedColumnPage.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/blocklet/BlockletEncodedColumnPage.java b/core/src/main/java/org/apache/carbondata/core/datastore/blocklet/BlockletEncodedColumnPage.java index 15adb2d..8abc0e4 100644 --- a/core/src/main/java/org/apache/carbondata/core/datastore/blocklet/BlockletEncodedColumnPage.java +++ b/core/src/main/java/org/apache/carbondata/core/datastore/blocklet/BlockletEncodedColumnPage.java @@ -123,7 +123,7 @@ public class BlockletEncodedColumnPage { else { isLocalDictEncoded = false; pageLevelDictionary = null; - LOGGER.warn("Local dictionary Fallback is initiated for column: " + this.columnName + LOGGER.info("Local dictionary Fallback is initiated for column: " + this.columnName + " for pages: 1 to " + encodedColumnPageList.size()); // submit all the older pages encoded with dictionary for fallback for (int pageIndex = 0; pageIndex < encodedColumnPageList.size(); pageIndex++) { http://git-wip-us.apache.org/repos/asf/carbondata/blob/36e14e51/core/src/main/java/org/apache/carbondata/core/datastore/page/LocalDictColumnPage.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/LocalDictColumnPage.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/LocalDictColumnPage.java index a072852..94e56b8 100644 --- a/core/src/main/java/org/apache/carbondata/core/datastore/page/LocalDictColumnPage.java +++ b/core/src/main/java/org/apache/carbondata/core/datastore/page/LocalDictColumnPage.java @@ -122,8 +122,8 @@ public class LocalDictColumnPage extends ColumnPage { dummyKey[0] = pageLevelDictionary.getDictionaryValue(bytes); encodedDataColumnPage.putBytes(rowId, keyGenerator.generateKey(dummyKey)); } catch (DictionaryThresholdReachedException e) { - LOGGER.error(e, "Local Dictionary threshold reached for the column: " + actualDataColumnPage - .getColumnSpec().getFieldName()); + LOGGER.warn("Local Dictionary threshold reached for the column: " + actualDataColumnPage + .getColumnSpec().getFieldName() + ", " + e.getMessage()); pageLevelDictionary = null; encodedDataColumnPage.freeMemory(); encodedDataColumnPage = null; http://git-wip-us.apache.org/repos/asf/carbondata/blob/36e14e51/core/src/main/java/org/apache/carbondata/core/localdictionary/generator/ColumnLocalDictionaryGenerator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/localdictionary/generator/ColumnLocalDictionaryGenerator.java b/core/src/main/java/org/apache/carbondata/core/localdictionary/generator/ColumnLocalDictionaryGenerator.java index 02ecb7f..b0c7275 100644 --- a/core/src/main/java/org/apache/carbondata/core/localdictionary/generator/ColumnLocalDictionaryGenerator.java +++ b/core/src/main/java/org/apache/carbondata/core/localdictionary/generator/ColumnLocalDictionaryGenerator.java @@ -67,7 +67,7 @@ public class ColumnLocalDictionaryGenerator implements LocalDictionaryGenerator currentSize += data.length; if (currentSize >= Integer.MAX_VALUE) { throw new DictionaryThresholdReachedException( - "Unable to generate dictionary as Dictionary Size crossed 2GB limit"); + "Unable to generate dictionary. Dictionary Size crossed 2GB limit"); } return this.dictionaryHolder.putIfAbsent(data); } http://git-wip-us.apache.org/repos/asf/carbondata/blob/36e14e51/core/src/test/java/org/apache/carbondata/core/localdictionary/TestPageLevelDictionary.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/carbondata/core/localdictionary/TestPageLevelDictionary.java b/core/src/test/java/org/apache/carbondata/core/localdictionary/TestPageLevelDictionary.java index a2a25bd..3337a7d 100644 --- a/core/src/test/java/org/apache/carbondata/core/localdictionary/TestPageLevelDictionary.java +++ b/core/src/test/java/org/apache/carbondata/core/localdictionary/TestPageLevelDictionary.java @@ -1,3 +1,19 @@ +/* + * 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.localdictionary; import java.io.IOException; http://git-wip-us.apache.org/repos/asf/carbondata/blob/36e14e51/integration/spark-common-test/src/test/resources/localdictionary.csv ---------------------------------------------------------------------- diff --git a/integration/spark-common-test/src/test/resources/localdictionary.csv b/integration/spark-common-test/src/test/resources/localdictionary.csv new file mode 100644 index 0000000..9f8adc7 --- /dev/null +++ b/integration/spark-common-test/src/test/resources/localdictionary.csv @@ -0,0 +1,8 @@ +name,age +vishal,30 +akash,24 +praveen,22 +kumar,30 +brijoo,35 +ravindra,34 + http://git-wip-us.apache.org/repos/asf/carbondata/blob/36e14e51/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/localdictionary/LocalDictionarySupportLoadTableTest.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/localdictionary/LocalDictionarySupportLoadTableTest.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/localdictionary/LocalDictionarySupportLoadTableTest.scala new file mode 100644 index 0000000..a3ab851 --- /dev/null +++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/localdictionary/LocalDictionarySupportLoadTableTest.scala @@ -0,0 +1,309 @@ +/* + * 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.testsuite.localdictionary + +import scala.collection.JavaConverters._ +import java.io.{File, PrintWriter} +import java.util +import java.util.Collections + +import org.apache.spark.sql.test.util.QueryTest +import org.scalatest.{BeforeAndAfterAll, Ignore} + +import org.apache.carbondata.core.cache.dictionary.DictionaryByteArrayWrapper +import org.apache.carbondata.core.constants.CarbonCommonConstants +import org.apache.carbondata.core.datastore.block.TableBlockInfo +import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk +import org.apache.carbondata.core.datastore.chunk.reader.CarbonDataReaderFactory +import org.apache.carbondata.core.datastore.chunk.reader.dimension.v3.CompressedDimensionChunkFileBasedReaderV3 +import org.apache.carbondata.core.datastore.compression.CompressorFactory +import org.apache.carbondata.core.datastore.filesystem.{CarbonFile, CarbonFileFilter} +import org.apache.carbondata.core.datastore.impl.FileFactory +import org.apache.carbondata.core.datastore.page.encoding.DefaultEncodingFactory +import org.apache.carbondata.core.metadata.ColumnarFormatVersion +import org.apache.carbondata.core.util.{CarbonProperties, DataFileFooterConverterV3} + +class LocalDictionarySupportLoadTableTest extends QueryTest with BeforeAndAfterAll { + + val file1 = resourcesPath + "/local_dictionary_source1.csv" + + val file2 = resourcesPath + "/local_dictionary_complex_data.csv" + + val storePath = warehouse + "/local2/Fact/Part0/Segment_0" + + override protected def beforeAll(): Unit = { + CarbonProperties.getInstance.addProperty(CarbonCommonConstants.BLOCKLET_SIZE, "10000") + createFile(file1) + createComplexData(file2) + sql("drop table if exists local2") + } + + test("test LocalDictionary Load For FallBackScenario"){ + sql("drop table if exists local2") + sql( + "CREATE TABLE local2(name string) STORED BY 'carbondata' tblproperties" + + "('local_dictionary_threshold'='2001','local_dictionary_include'='name')") + sql("load data inpath '" + file1 + "' into table local2 OPTIONS('header'='false')") + assert(!checkForLocalDictionary(getDimRawChunk(0))) + } + + test("test successful local dictionary generation"){ + sql("drop table if exists local2") + sql( + "CREATE TABLE local2(name string) STORED BY 'carbondata' tblproperties" + + "('local_dictionary_threshold'='9001','local_dictionary_include'='name')") + sql("load data inpath '" + file1 + "' into table local2 OPTIONS('header'='false')") + assert(checkForLocalDictionary(getDimRawChunk(0))) + } + + test("test successful local dictionary generation for default configs") { + sql("drop table if exists local2") + sql("CREATE TABLE local2(name string) STORED BY 'carbondata'") + sql("load data inpath '" + file1 + "' into table local2 OPTIONS('header'='false')") + assert(checkForLocalDictionary(getDimRawChunk(0))) + } + + test("test local dictionary generation for local dictioanry include") { + sql("drop table if exists local2") + sql( + "CREATE TABLE local2(name string) STORED BY 'carbondata' tblproperties" + + "('dictionary_include'='name')") + sql("load data inpath '" + file1 + "' into table local2 OPTIONS('header'='false')") + assert(!checkForLocalDictionary(getDimRawChunk(0))) + } + + test("test local dictionary generation for local dictioanry exclude"){ + sql("drop table if exists local2") + sql( + "CREATE TABLE local2(name string) STORED BY 'carbondata' tblproperties" + + "('dictionary_exclude'='name')") + sql("load data inpath '" + file1 + "' into table local2 OPTIONS('header'='false')") + assert(checkForLocalDictionary(getDimRawChunk(0))) + } + + test("test local dictionary generation when it is disabled"){ + sql("drop table if exists local2") + sql( + "CREATE TABLE local2(name string) STORED BY 'carbondata' tblproperties" + + "('local_dictionary_enable'='false','local_dictionary_include'='name')") + sql("load data inpath '" + file1 + "' into table local2 OPTIONS('header'='false')") + assert(!checkForLocalDictionary(getDimRawChunk(0))) + } + + test("test local dictionary generation for invalid threshold configurations"){ + sql("drop table if exists local2") + sql( + "CREATE TABLE local2(name string) STORED BY 'carbondata' tblproperties" + + "('local_dictionary_include'='name','local_dictionary_threshold'='300000')") + sql("load data inpath '" + file1 + "' into table local2 OPTIONS('header'='false')") + assert(checkForLocalDictionary(getDimRawChunk(0))) + } + + test("test local dictionary generation for include and exclude"){ + sql("drop table if exists local2") + sql( + "CREATE TABLE local2(name string, age string) STORED BY 'carbondata' tblproperties" + + "('local_dictionary_include'='name', 'local_dictionary_exclude'='age')") + sql("insert into table local2 values('vishal', '30')") + assert(checkForLocalDictionary(getDimRawChunk(0))) + assert(!checkForLocalDictionary(getDimRawChunk(1))) + } + + test("test local dictionary generation for complex type"){ + sql("drop table if exists local2") + sql( + "CREATE TABLE local2(name struct<i:string,s:string>) STORED BY 'carbondata' tblproperties" + + "('local_dictionary_include'='name')") + sql("load data inpath '" + file2 + + "' into table local2 OPTIONS('header'='false','COMPLEX_DELIMITER_LEVEL_1'='$', " + + "'COMPLEX_DELIMITER_LEVEL_2'=':')") + assert(!checkForLocalDictionary(getDimRawChunk(0))) + assert(checkForLocalDictionary(getDimRawChunk(1))) + assert(checkForLocalDictionary(getDimRawChunk(2))) + } + + test("test to validate local dictionary values"){ + sql("drop table if exists local2") + sql("CREATE TABLE local2(name string) STORED BY 'carbondata'") + sql("load data inpath '" + resourcesPath + "/localdictionary.csv" + "' into table local2") + val dimRawChunk = getDimRawChunk(0) + val dictionaryData = Array("vishal", "kumar", "akash", "praveen", "brijoo") + try + assert(validateDictionary(dimRawChunk.get(0), dictionaryData)) + catch { + case e: Exception => + assert(false) + } + } + + override protected def afterAll(): Unit = { + sql("drop table if exists local2") + deleteFile(file1) + deleteFile(file2) + CarbonProperties.getInstance + .addProperty(CarbonCommonConstants.BLOCKLET_SIZE, + CarbonCommonConstants.BLOCKLET_SIZE_DEFAULT_VAL) + } + + /** + * create the csv data file + * + * @param fileName + */ + private def createFile(fileName: String, line: Int = 9000, start: Int = 0): Unit = { + val writer = new PrintWriter(new File(fileName)) + val data: util.ArrayList[String] = new util.ArrayList[String]() + for (i <- start until line) { + data.add("n" + i) + } + Collections.sort(data) + data.asScala.foreach { eachdata => + writer.println(eachdata) + } + writer.close() + } + + /** + * create the csv for complex data + * + * @param fileName + */ + private def createComplexData(fileName: String, line: Int = 9000, start: Int = 0): Unit = { + val writer = new PrintWriter(new File(fileName)) + val data: util.ArrayList[String] = new util.ArrayList[String]() + for (i <- start until line) { + data.add("n" + i + "$" + (i + 1)) + } + Collections.sort(data) + data.asScala.foreach { eachdata => + writer.println(eachdata) + } + writer.close() + } + + /** + * delete csv file after test execution + * + * @param fileName + */ + private def deleteFile(fileName: String): Unit = { + val file: File = new File(fileName) + if (file.exists) file.delete + } + + private def checkForLocalDictionary(dimensionRawColumnChunks: util + .List[DimensionRawColumnChunk]): Boolean = { + var isLocalDictionaryGenerated = false + import scala.collection.JavaConversions._ + for (dimensionRawColumnChunk <- dimensionRawColumnChunks) { + if (dimensionRawColumnChunk.getDataChunkV3 + .isSetLocal_dictionary) { + isLocalDictionaryGenerated = true + } + } + isLocalDictionaryGenerated + } + + /** + * this method returns true if local dictionary is created for all the blocklets or not + * + * @return + */ + private def getDimRawChunk(blockindex: Int): util.ArrayList[DimensionRawColumnChunk] = { + val dataFiles = FileFactory.getCarbonFile(storePath) + .listFiles(new CarbonFileFilter() { + override def accept(file: CarbonFile): Boolean = { + if (file.getName + .endsWith(CarbonCommonConstants.FACT_FILE_EXT)) { + true + } else { + false + } + } + }) + val dimensionRawColumnChunks = read(dataFiles(0).getAbsolutePath, + blockindex) + dimensionRawColumnChunks + } + + private def read(filePath: String, blockIndex: Int) = { + val carbonDataFiles = new File(filePath) + val dimensionRawColumnChunks = new + util.ArrayList[DimensionRawColumnChunk] + val offset = carbonDataFiles.length + val converter = new DataFileFooterConverterV3 + val fileReader = FileFactory.getFileHolder(FileFactory.getFileType(filePath)) + val actualOffset = fileReader.readLong(carbonDataFiles.getAbsolutePath, offset - 8) + val blockInfo = new TableBlockInfo(carbonDataFiles.getAbsolutePath, + actualOffset, + "0", + new Array[String](0), + carbonDataFiles.length, + ColumnarFormatVersion.V3, + null) + val dataFileFooter = converter.readDataFileFooter(blockInfo) + val blockletList = dataFileFooter.getBlockletList + import scala.collection.JavaConversions._ + for (blockletInfo <- blockletList) { + val dimensionColumnChunkReader = + CarbonDataReaderFactory + .getInstance + .getDimensionColumnChunkReader(ColumnarFormatVersion.V3, + blockletInfo, + dataFileFooter.getSegmentInfo.getColumnCardinality, + carbonDataFiles.getAbsolutePath, + false).asInstanceOf[CompressedDimensionChunkFileBasedReaderV3] + dimensionRawColumnChunks + .add(dimensionColumnChunkReader.readRawDimensionChunk(fileReader, blockIndex)) + } + dimensionRawColumnChunks + } + + private def validateDictionary(rawColumnPage: DimensionRawColumnChunk, + data: Array[String]): Boolean = { + val local_dictionary = rawColumnPage.getDataChunkV3.local_dictionary + if (null != local_dictionary) { + val encodings = local_dictionary.getDictionary_meta.encoders + val encoderMetas = local_dictionary.getDictionary_meta.getEncoder_meta + val encodingFactory = DefaultEncodingFactory.getInstance + val decoder = encodingFactory.createDecoder(encodings, encoderMetas) + val dictionaryPage = decoder + .decode(local_dictionary.getDictionary_data, 0, local_dictionary.getDictionary_data.length) + val dictionaryMap = new + util.HashMap[DictionaryByteArrayWrapper, Integer] + val usedDictionaryValues = util.BitSet + .valueOf(CompressorFactory.getInstance.getCompressor + .unCompressByte(local_dictionary.getDictionary_values)) + var index = 0 + var i = usedDictionaryValues.nextSetBit(0) + while ( { i >= 0 }) { + dictionaryMap + .put(new DictionaryByteArrayWrapper(dictionaryPage.getBytes({ index += 1; index - 1 })), + i) + i = usedDictionaryValues.nextSetBit(i + 1) + } + for (i <- data.indices) { + if (null == dictionaryMap.get(new DictionaryByteArrayWrapper(data(i).getBytes))) { + return false + } + } + return true + } + false + } + +}
