Repository: carbondata Updated Branches: refs/heads/master 459331c32 -> 8cb37dd75
[CARBONDATA-2644][DataLoad]ADD carbon.load.sortMemory.spill.percentage parameter invalid value check This closes #2397 Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/8cb37dd7 Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/8cb37dd7 Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/8cb37dd7 Branch: refs/heads/master Commit: 8cb37dd75541e23fbb0cfabd1f853d45ea51e5c2 Parents: 459331c Author: ndwangsen <[email protected]> Authored: Fri Jun 22 12:02:36 2018 +0800 Committer: Jacky Li <[email protected]> Committed: Sat Jun 30 23:04:30 2018 +0800 ---------------------------------------------------------------------- .../constants/CarbonLoadOptionConstants.java | 6 +-- .../carbondata/core/util/CarbonProperties.java | 50 ++++++++++++++++++++ .../dataload/TestLoadDataWithUnsafeMemory.scala | 10 +++- .../unsafe/merger/UnsafeIntermediateMerger.java | 13 +---- 4 files changed, 63 insertions(+), 16 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/carbondata/blob/8cb37dd7/core/src/main/java/org/apache/carbondata/core/constants/CarbonLoadOptionConstants.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/constants/CarbonLoadOptionConstants.java b/core/src/main/java/org/apache/carbondata/core/constants/CarbonLoadOptionConstants.java index 6b8281c..3eab69d 100644 --- a/core/src/main/java/org/apache/carbondata/core/constants/CarbonLoadOptionConstants.java +++ b/core/src/main/java/org/apache/carbondata/core/constants/CarbonLoadOptionConstants.java @@ -144,12 +144,12 @@ public final class CarbonLoadOptionConstants { * If the sort memory is insufficient, spill inmemory pages to disk. * The total amount of pages is at most the specified percentage of total sort memory. Default * value 0 means that no pages will be spilled and the newly incoming pages will be spilled, - * whereas value 1 means that all pages will be spilled and newly incoming pages will be loaded - * into sort memory. + * whereas value 100 means that all pages will be spilled and newly incoming pages will be loaded + * into sort memory, valid value is from 0 to 100. */ @CarbonProperty public static final String CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE - = "carbon.load.sortMemory.spill.percentage"; + = "carbon.load.sortmemory.spill.percentage"; public static final String CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE_DEFAULT = "0"; /** http://git-wip-us.apache.org/repos/asf/carbondata/blob/8cb37dd7/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java index dc50ab0..4df10ee 100644 --- a/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java +++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java @@ -65,6 +65,7 @@ import static org.apache.carbondata.core.constants.CarbonCommonConstants.NUM_COR import static org.apache.carbondata.core.constants.CarbonCommonConstants.NUM_CORES_BLOCK_SORT; import static org.apache.carbondata.core.constants.CarbonCommonConstants.SORT_INTERMEDIATE_FILES_LIMIT; import static org.apache.carbondata.core.constants.CarbonCommonConstants.SORT_SIZE; +import static org.apache.carbondata.core.constants.CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE; import static org.apache.carbondata.core.constants.CarbonV3DataFormatConstants.BLOCKLET_SIZE_IN_MB; import static org.apache.carbondata.core.constants.CarbonV3DataFormatConstants.NUMBER_OF_COLUMN_TO_READ_IN_IO; @@ -197,6 +198,9 @@ public final class CarbonProperties { case CARBON_SEARCH_MODE_WORKER_WORKLOAD_LIMIT: validatePositiveInteger(CARBON_SEARCH_MODE_WORKER_WORKLOAD_LIMIT); break; + case CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE: + validateSortMemorySpillPercentage(); + break; // TODO : Validation for carbon.lock.type should be handled for addProperty flow default: // none @@ -262,6 +266,7 @@ public final class CarbonProperties { validateWorkingMemory(); validateSortStorageMemory(); validateEnableQueryStatistics(); + validateSortMemorySpillPercentage(); } /** @@ -1599,4 +1604,49 @@ public final class CarbonProperties { } return storageLevel.toUpperCase(); } + + /** + * Return valid storage level for CARBON_INSERT_STORAGE_LEVEL + * @return String + */ + public int getSortMemorySpillPercentage() { + int spillPercentage = 0; + try { + String spillPercentageStr = getProperty( + CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE, + CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE_DEFAULT); + spillPercentage = Integer.parseInt(spillPercentageStr); + } catch (NumberFormatException e) { + spillPercentage = Integer.parseInt( + CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE_DEFAULT); + } + return spillPercentage; + } + + private void validateSortMemorySpillPercentage() { + String spillPercentageStr = carbonProperties.getProperty( + CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE, + CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE_DEFAULT); + + try { + int spillPercentage = Integer.parseInt(spillPercentageStr); + if (spillPercentage > 100 || spillPercentage < 0) { + LOGGER.info( + "The sort memory spill percentage value \"" + spillPercentageStr + + "\" is invalid. Using the default value \"" + + CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE_DEFAULT); + carbonProperties.setProperty( + CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE, + CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE_DEFAULT); + } + } catch (NumberFormatException e) { + LOGGER.info( + "The sort memory spill percentage value \"" + spillPercentageStr + + "\" is invalid. Using the default value \"" + + CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE_DEFAULT); + carbonProperties.setProperty( + CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE, + CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE_DEFAULT); + } + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/8cb37dd7/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataWithUnsafeMemory.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataWithUnsafeMemory.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataWithUnsafeMemory.scala index c08ec52..0545f05 100644 --- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataWithUnsafeMemory.scala +++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataWithUnsafeMemory.scala @@ -21,7 +21,7 @@ import org.apache.spark.sql.Row import org.apache.spark.sql.test.util.QueryTest import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Ignore} -import org.apache.carbondata.core.constants.CarbonCommonConstants +import org.apache.carbondata.core.constants.{CarbonCommonConstants, CarbonLoadOptionConstants} import org.apache.carbondata.core.util.CarbonProperties /** @@ -44,6 +44,9 @@ class TestLoadDataWithUnsafeMemory extends QueryTest val originUnsafeSizeForChunk: String = CarbonProperties.getInstance() .getProperty(CarbonCommonConstants.OFFHEAP_SORT_CHUNK_SIZE_IN_MB, CarbonCommonConstants.OFFHEAP_SORT_CHUNK_SIZE_IN_MB_DEFAULT) + val originSpillPercentage: String = CarbonProperties.getInstance() + .getProperty(CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE, + CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE_DEFAULT) val targetTable = "table_unsafe_memory" @@ -64,6 +67,8 @@ class TestLoadDataWithUnsafeMemory extends QueryTest .addProperty(CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB, "512") CarbonProperties.getInstance() .addProperty(CarbonCommonConstants.OFFHEAP_SORT_CHUNK_SIZE_IN_MB, "512") + CarbonProperties.getInstance() + .addProperty(CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE, "-1") } override def afterAll(): Unit = { @@ -75,6 +80,9 @@ class TestLoadDataWithUnsafeMemory extends QueryTest .addProperty(CarbonCommonConstants.UNSAFE_WORKING_MEMORY_IN_MB, originUnsafeMemForWorking) CarbonProperties.getInstance() .addProperty(CarbonCommonConstants.OFFHEAP_SORT_CHUNK_SIZE_IN_MB, originUnsafeSizeForChunk) + CarbonProperties.getInstance() + .addProperty(CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE, + originSpillPercentage) } private def testSimpleTable(): Unit = { http://git-wip-us.apache.org/repos/asf/carbondata/blob/8cb37dd7/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/merger/UnsafeIntermediateMerger.java ---------------------------------------------------------------------- diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/merger/UnsafeIntermediateMerger.java b/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/merger/UnsafeIntermediateMerger.java index 8d2c52a..bd2f5d9 100644 --- a/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/merger/UnsafeIntermediateMerger.java +++ b/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/merger/UnsafeIntermediateMerger.java @@ -30,7 +30,6 @@ import java.util.concurrent.TimeUnit; 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.constants.CarbonLoadOptionConstants; import org.apache.carbondata.core.memory.UnsafeSortMemoryManager; import org.apache.carbondata.core.util.CarbonProperties; import org.apache.carbondata.core.util.CarbonThreadFactory; @@ -82,17 +81,7 @@ public class UnsafeIntermediateMerger { SynchronizedList.decorate(new ArrayList<File>(CarbonCommonConstants.CONSTANT_SIZE_TEN)); this.mergerTask = new ArrayList<>(); - Integer spillPercentage; - try { - String spillPercentageStr = CarbonProperties.getInstance().getProperty( - CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE, - CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE_DEFAULT); - spillPercentage = Integer.valueOf(spillPercentageStr); - } catch (NumberFormatException e) { - spillPercentage = Integer.valueOf( - CarbonLoadOptionConstants.CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE_DEFAULT); - } - + Integer spillPercentage = CarbonProperties.getInstance().getSortMemorySpillPercentage(); this.spillSizeInSortMemory = UnsafeSortMemoryManager.INSTANCE.getUsableMemory() * spillPercentage / 100; }
