Repository: carbondata Updated Branches: refs/heads/master c3bc1ba10 -> 7a1d12aa1
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7a1d12aa/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModelBuilder.java ---------------------------------------------------------------------- diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModelBuilder.java b/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModelBuilder.java index a88ce60..0b60fa5 100644 --- a/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModelBuilder.java +++ b/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModelBuilder.java @@ -37,6 +37,7 @@ import org.apache.carbondata.core.util.CarbonUtil; import org.apache.carbondata.processing.loading.constants.DataLoadProcessorConstants; import org.apache.carbondata.processing.loading.csvinput.CSVInputFormat; import org.apache.carbondata.processing.loading.sort.SortScopeOptions; +import org.apache.carbondata.processing.util.CarbonBadRecordUtil; import org.apache.carbondata.processing.util.TableOptionConstant; import org.apache.commons.lang.StringUtils; @@ -60,7 +61,7 @@ public class CarbonLoadModelBuilder { * @param taskNo * @return a new CarbonLoadModel instance */ - public CarbonLoadModel build(Map<String, String> options, long UUID, String taskNo) + public CarbonLoadModel build(Map<String, String> options, long UUID, String taskNo) throws InvalidLoadOptionException, IOException { Map<String, String> optionsFinal = LoadOption.fillOptionWithDefaultValue(options); @@ -72,6 +73,7 @@ public class CarbonLoadModelBuilder { } optionsFinal.put("fileheader", Strings.mkString(columns, ",")); } + optionsFinal.put("bad_record_path", CarbonBadRecordUtil.getBadRecordsPath(options, table)); CarbonLoadModel model = new CarbonLoadModel(); model.setCarbonTransactionalTable(table.isTransactionalTable()); model.setFactTimeStamp(UUID); @@ -163,10 +165,12 @@ public class CarbonLoadModelBuilder { if (Boolean.parseBoolean(bad_records_logger_enable) || LoggerAction.REDIRECT.name().equalsIgnoreCase(bad_records_action)) { - if (!CarbonUtil.isValidBadStorePath(bad_record_path)) { - throw new InvalidLoadOptionException("Invalid bad records location."); + if (!StringUtils.isEmpty(bad_record_path)) { + bad_record_path = CarbonUtil.checkAndAppendHDFSUrl(bad_record_path); + } else { + throw new InvalidLoadOptionException( + "Cannot redirect bad records as bad record location is not provided."); } - bad_record_path = CarbonUtil.checkAndAppendHDFSUrl(bad_record_path); } carbonLoadModel.setBadRecordsLocation(bad_record_path); http://git-wip-us.apache.org/repos/asf/carbondata/blob/7a1d12aa/processing/src/main/java/org/apache/carbondata/processing/loading/model/LoadOption.java ---------------------------------------------------------------------- diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/model/LoadOption.java b/processing/src/main/java/org/apache/carbondata/processing/loading/model/LoadOption.java index 17c3651..9733816 100644 --- a/processing/src/main/java/org/apache/carbondata/processing/loading/model/LoadOption.java +++ b/processing/src/main/java/org/apache/carbondata/processing/loading/model/LoadOption.java @@ -159,17 +159,6 @@ public class LoadOption { CarbonCommonConstants.LOAD_BATCH_SORT_SIZE_INMB, CarbonCommonConstants.LOAD_BATCH_SORT_SIZE_INMB_DEFAULT)))); - optionsFinal.put( - "bad_record_path", - Maps.getOrDefault( - options, - "bad_record_path", - CarbonProperties.getInstance().getProperty( - CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORD_PATH, - CarbonProperties.getInstance().getProperty( - CarbonCommonConstants.CARBON_BADRECORDS_LOC, - CarbonCommonConstants.CARBON_BADRECORDS_LOC_DEFAULT_VAL)))); - String useOnePass = Maps.getOrDefault( options, "single_pass", http://git-wip-us.apache.org/repos/asf/carbondata/blob/7a1d12aa/processing/src/main/java/org/apache/carbondata/processing/util/CarbonBadRecordUtil.java ---------------------------------------------------------------------- diff --git a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonBadRecordUtil.java b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonBadRecordUtil.java index c494eef..55bc580 100644 --- a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonBadRecordUtil.java +++ b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonBadRecordUtil.java @@ -19,6 +19,7 @@ package org.apache.carbondata.processing.util; import java.io.File; import java.io.IOException; +import java.util.Map; import org.apache.carbondata.common.logging.LogService; import org.apache.carbondata.common.logging.LogServiceFactory; @@ -27,12 +28,14 @@ import org.apache.carbondata.core.constants.CarbonLoadOptionConstants; import org.apache.carbondata.core.datastore.filesystem.CarbonFile; import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter; import org.apache.carbondata.core.datastore.impl.FileFactory; -import org.apache.carbondata.core.metadata.CarbonTableIdentifier; +import org.apache.carbondata.core.metadata.schema.table.CarbonTable; import org.apache.carbondata.core.util.CarbonProperties; import org.apache.carbondata.processing.loading.BadRecordsLogger; import org.apache.carbondata.processing.loading.CarbonDataLoadConfiguration; import org.apache.carbondata.processing.loading.model.CarbonLoadModel; +import org.apache.commons.lang.StringUtils; + /** * Common methods used for the bad record handling */ @@ -47,14 +50,11 @@ public class CarbonBadRecordUtil { */ public static void renameBadRecord(CarbonDataLoadConfiguration configuration) { // rename the bad record in progress to normal - CarbonTableIdentifier identifier = - configuration.getTableIdentifier().getCarbonTableIdentifier(); String storeLocation = ""; if (configuration.isCarbonTransactionalTable()) { storeLocation = - identifier.getDatabaseName() + CarbonCommonConstants.FILE_SEPARATOR + identifier - .getTableName() + CarbonCommonConstants.FILE_SEPARATOR + configuration.getSegmentId() - + CarbonCommonConstants.FILE_SEPARATOR + configuration.getTaskNo(); + configuration.getSegmentId() + CarbonCommonConstants.FILE_SEPARATOR + configuration + .getTaskNo(); } else { storeLocation = "SdkWriterBadRecords" + CarbonCommonConstants.FILE_SEPARATOR + configuration.getTaskNo(); @@ -127,4 +127,31 @@ public class CarbonBadRecordUtil { return (null != BadRecordsLogger.hasBadRecord(key)); } + public static String getBadRecordsPath(Map<String, String> loadOptions, CarbonTable table) { + String badRecordsFromLoad = loadOptions.get("bad_record_path"); + String badRecordsFromCreate = + table.getTableInfo().getFactTable().getTableProperties().get("bad_records_path"); + String badRecordsPath; + if (StringUtils.isNotEmpty(badRecordsFromLoad)) { + badRecordsPath = + badRecordsFromLoad + CarbonCommonConstants.FILE_SEPARATOR + table.getDatabaseName() + + CarbonCommonConstants.FILE_SEPARATOR + table.getTableName(); + } else if (StringUtils.isNotEmpty(badRecordsFromCreate)) { + badRecordsPath = badRecordsFromCreate; + } else { + String badRecordsFromProp = CarbonProperties.getInstance() + .getProperty(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORD_PATH, + CarbonProperties.getInstance() + .getProperty(CarbonCommonConstants.CARBON_BADRECORDS_LOC, + CarbonCommonConstants.CARBON_BADRECORDS_LOC_DEFAULT_VAL)); + if (!badRecordsFromProp.isEmpty()) { + badRecordsFromProp = + badRecordsFromProp + CarbonCommonConstants.FILE_SEPARATOR + table.getDatabaseName() + + CarbonCommonConstants.FILE_SEPARATOR + table.getTableName(); + } + badRecordsPath = badRecordsFromProp; + } + return badRecordsPath; + } + }
