carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jack...@apache.org
Subject [1/2] incubator-carbondata git commit: [CARBONDATA-288] In hdfs bad record logger is failing in writing the bad records
Date Fri, 14 Oct 2016 14:09:42 GMT
Repository: incubator-carbondata
Updated Branches:
  refs/heads/master 7ea31a6ae -> 5b978f5b7


[CARBONDATA-288] In hdfs bad record logger is failing in writing the bad records


Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/809a4d00
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/809a4d00
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/809a4d00

Branch: refs/heads/master
Commit: 809a4d00976bad33002a4f4e32cac082d2e08c4f
Parents: 7ea31a6
Author: mohammadshahidkhan <mohdshahidkhan1987@gmail.com>
Authored: Sun Oct 9 02:24:57 2016 +0530
Committer: jackylk <jacky.likun@huawei.com>
Committed: Fri Oct 14 22:07:51 2016 +0800

----------------------------------------------------------------------
 .../hadoop/test/util/StoreCreator.java          | 10 +++--
 .../carbondata/spark/load/CarbonLoadModel.java  | 22 +++++-----
 .../carbondata/spark/load/CarbonLoaderUtil.java |  2 +-
 .../execution/command/carbonTableSchema.scala   | 12 ++++--
 .../processing/api/dataloader/SchemaInfo.java   | 18 ++++----
 .../processing/constants/LoggerAction.java      | 19 +++++++++
 .../constants/TableOptionConstant.java          | 43 ++++++++++++++++++++
 .../graphgenerator/GraphGenerator.java          |  2 +-
 .../csvbased/BadRecordslogger.java              | 40 ++++++++++--------
 .../csvbased/CarbonCSVBasedSeqGenStep.java      | 16 +++++---
 10 files changed, 132 insertions(+), 52 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/809a4d00/hadoop/src/test/java/org/apache/carbondata/hadoop/test/util/StoreCreator.java
----------------------------------------------------------------------
diff --git a/hadoop/src/test/java/org/apache/carbondata/hadoop/test/util/StoreCreator.java
b/hadoop/src/test/java/org/apache/carbondata/hadoop/test/util/StoreCreator.java
index 5661888..7c9e170 100644
--- a/hadoop/src/test/java/org/apache/carbondata/hadoop/test/util/StoreCreator.java
+++ b/hadoop/src/test/java/org/apache/carbondata/hadoop/test/util/StoreCreator.java
@@ -62,6 +62,7 @@ import org.apache.carbondata.lcm.fileoperations.AtomicFileOperationsImpl;
 import org.apache.carbondata.lcm.fileoperations.FileWriteOperation;
 import org.apache.carbondata.processing.api.dataloader.DataLoadModel;
 import org.apache.carbondata.processing.api.dataloader.SchemaInfo;
+import org.apache.carbondata.processing.constants.TableOptionConstant;
 import org.apache.carbondata.processing.csvload.DataGraphExecuter;
 import org.apache.carbondata.processing.dataprocessor.DataProcessTaskStatus;
 import org.apache.carbondata.processing.dataprocessor.IDataProcessStatus;
@@ -358,9 +359,12 @@ public class StoreCreator {
     schmaModel.setCommentCharacter("#");
     info.setDatabaseName(databaseName);
     info.setTableName(tableName);
-    info.setSerializationNullFormat("serialization_null_format" + "," + "\\N");
-    info.setBadRecordsLoggerEnable("bad_records_logger_enable"+","+"false");
-    info.setBadRecordsLoggerEnable("bad_records_action"+","+"force");
+    info.setSerializationNullFormat(
+        TableOptionConstant.SERIALIZATION_NULL_FORMAT.getName() + "," + "\\N");
+    info.setBadRecordsLoggerEnable(
+        TableOptionConstant.BAD_RECORDS_LOGGER_ENABLE.getName() + "," + "false");
+    info.setBadRecordsLoggerAction(
+        TableOptionConstant.BAD_RECORDS_ACTION.getName() + "," + "force");
 
     generateGraph(schmaModel, info, loadModel.getTableName(), "0", loadModel.getSchema(),
null,
         loadModel.getLoadMetadataDetails());

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/809a4d00/integration/spark/src/main/java/org/apache/carbondata/spark/load/CarbonLoadModel.java
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/java/org/apache/carbondata/spark/load/CarbonLoadModel.java
b/integration/spark/src/main/java/org/apache/carbondata/spark/load/CarbonLoadModel.java
index 3fd481b..106ad71 100644
--- a/integration/spark/src/main/java/org/apache/carbondata/spark/load/CarbonLoadModel.java
+++ b/integration/spark/src/main/java/org/apache/carbondata/spark/load/CarbonLoadModel.java
@@ -117,9 +117,9 @@ public class CarbonLoadModel implements Serializable {
   private String badRecordsLoggerEnable;
 
   /**
-   * defines the option to specify the bad record log redirect to raw csv
+   * defines the option to specify the bad record logger action
    */
-  private String badRecordsLoggerRedirect;
+  private String badRecordsAction;
 
   /**
    * Max number of columns that needs to be parsed by univocity parser
@@ -348,7 +348,7 @@ public class CarbonLoadModel implements Serializable {
     copy.segmentId = segmentId;
     copy.serializationNullFormat = serializationNullFormat;
     copy.badRecordsLoggerEnable = badRecordsLoggerEnable;
-    copy.badRecordsLoggerRedirect =badRecordsLoggerRedirect;
+    copy.badRecordsAction = badRecordsAction;
     copy.escapeChar = escapeChar;
     copy.quoteChar = quoteChar;
     copy.commentChar = commentChar;
@@ -391,7 +391,7 @@ public class CarbonLoadModel implements Serializable {
     copyObj.segmentId = segmentId;
     copyObj.serializationNullFormat = serializationNullFormat;
     copyObj.badRecordsLoggerEnable = badRecordsLoggerEnable;
-    copyObj.badRecordsLoggerRedirect =badRecordsLoggerRedirect;
+    copyObj.badRecordsAction = badRecordsAction;
     copyObj.escapeChar = escapeChar;
     copyObj.quoteChar = quoteChar;
     copyObj.commentChar = commentChar;
@@ -612,19 +612,19 @@ public class CarbonLoadModel implements Serializable {
   }
 
   /**
-   *  returns option to specify the bad record log redirect to raw csv
+   *  returns option to specify the bad record logger action
    * @return
    */
-  public String getBadRecordsLoggerRedirect() {
-    return badRecordsLoggerRedirect;
+  public String getBadRecordsAction() {
+    return badRecordsAction;
   }
 
   /**
-   * set option to specify the bad record log redirect to raw csv
-   * @param badRecordsLoggerRedirect
+   * set option to specify the bad record logger action
+   * @param badRecordsAction
    */
-  public void setBadRecordsLoggerRedirect(String badRecordsLoggerRedirect) {
-    this.badRecordsLoggerRedirect = badRecordsLoggerRedirect;
+  public void setBadRecordsAction(String badRecordsAction) {
+    this.badRecordsAction = badRecordsAction;
   }
 
   public String getRddIteratorKey() {

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/809a4d00/integration/spark/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java
b/integration/spark/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java
index 18c9538..8bc5fdc 100644
--- a/integration/spark/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java
+++ b/integration/spark/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java
@@ -198,7 +198,7 @@ public final class CarbonLoaderUtil {
     info.setComplexDelimiterLevel2(loadModel.getComplexDelimiterLevel2());
     info.setSerializationNullFormat(loadModel.getSerializationNullFormat());
     info.setBadRecordsLoggerEnable(loadModel.getBadRecordsLoggerEnable());
-    info.setBadRecordsLoggerRedirect(loadModel.getBadRecordsLoggerRedirect());
+    info.setBadRecordsLoggerAction(loadModel.getBadRecordsAction());
 
     generateGraph(schmaModel, info, loadModel, outPutLoc);
 

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/809a4d00/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
index 6dc6668..7b6a213 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
@@ -54,6 +54,7 @@ import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
 import org.apache.carbondata.integration.spark.merger.CompactionType
 import org.apache.carbondata.lcm.locks.{CarbonLockFactory, LockUsage}
 import org.apache.carbondata.lcm.status.SegmentStatusManager
+import org.apache.carbondata.processing.constants.TableOptionConstant
 import org.apache.carbondata.processing.etl.DataLoadingException
 import org.apache.carbondata.spark.CarbonSparkFactory
 import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
@@ -1135,12 +1136,15 @@ case class LoadTable(
       carbonLoadModel.setEscapeChar(escapeChar)
       carbonLoadModel.setQuoteChar(quoteChar)
       carbonLoadModel.setCommentChar(commentchar)
-      carbonLoadModel.setSerializationNullFormat("serialization_null_format" + "," +
-        serializationNullFormat)
       carbonLoadModel
-        .setBadRecordsLoggerEnable("bad_records_logger_enable" + "," + badRecordsLoggerEnable)
+        .setSerializationNullFormat(
+          TableOptionConstant.SERIALIZATION_NULL_FORMAT.getName + "," + serializationNullFormat)
       carbonLoadModel
-        .setBadRecordsLoggerRedirect("bad_records_action" + "," + badRecordsLoggerRedirect)
+        .setBadRecordsLoggerEnable(
+          TableOptionConstant.BAD_RECORDS_LOGGER_ENABLE.getName + "," + badRecordsLoggerEnable)
+      carbonLoadModel
+        .setBadRecordsAction(
+          TableOptionConstant.BAD_RECORDS_ACTION.getName + "," + badRecordsLoggerRedirect)
 
       if (delimiter.equalsIgnoreCase(complex_delimiter_level_1) ||
           complex_delimiter_level_1.equalsIgnoreCase(complex_delimiter_level_2) ||

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/809a4d00/processing/src/main/java/org/apache/carbondata/processing/api/dataloader/SchemaInfo.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/api/dataloader/SchemaInfo.java
b/processing/src/main/java/org/apache/carbondata/processing/api/dataloader/SchemaInfo.java
index 56de5ca..a00e913 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/api/dataloader/SchemaInfo.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/api/dataloader/SchemaInfo.java
@@ -69,9 +69,9 @@ public class SchemaInfo {
    */
   private String badRecordsLoggerEnable;
   /**
-   * defines the option to specify whether to redirect the bad record logger to raw csv or
not
+   * defines the option to specify whether to bad record logger action
    */
-  private String badRecordsLoggerRedirect;
+  private String badRecordsLoggerAction;
 
 
   public String getComplexDelimiterLevel1() {
@@ -215,18 +215,18 @@ public class SchemaInfo {
   }
 
   /**
-   * returns the option to set to redirect the badrecord logger to raw csv
+   * returns the option to set bad record logger action
    * @return
    */
-  public String getBadRecordsLoggerRedirect() {
-    return badRecordsLoggerRedirect;
+  public String getBadRecordsLoggerAction() {
+    return badRecordsLoggerAction;
   }
 
   /**
-   * set the option to set to redirect the badrecord logger to raw csv
-   * @param badRecordsLoggerRedirect
+   * set the option to set set bad record logger action
+   * @param badRecordsLoggerAction
    */
-  public void setBadRecordsLoggerRedirect(String badRecordsLoggerRedirect) {
-    this.badRecordsLoggerRedirect = badRecordsLoggerRedirect;
+  public void setBadRecordsLoggerAction(String badRecordsLoggerAction) {
+    this.badRecordsLoggerAction = badRecordsLoggerAction;
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/809a4d00/processing/src/main/java/org/apache/carbondata/processing/constants/LoggerAction.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/constants/LoggerAction.java
b/processing/src/main/java/org/apache/carbondata/processing/constants/LoggerAction.java
index bef65a9..622be42 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/constants/LoggerAction.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/constants/LoggerAction.java
@@ -1,3 +1,22 @@
+/*
+ * 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.processing.constants;
 
 /**

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/809a4d00/processing/src/main/java/org/apache/carbondata/processing/constants/TableOptionConstant.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/constants/TableOptionConstant.java
b/processing/src/main/java/org/apache/carbondata/processing/constants/TableOptionConstant.java
new file mode 100644
index 0000000..eef69f5
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/constants/TableOptionConstant.java
@@ -0,0 +1,43 @@
+/*
+ * 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.processing.constants;
+
+/**
+ * enum holds the value related to the ddl option
+ */
+public enum TableOptionConstant {
+  SERIALIZATION_NULL_FORMAT("serialization_null_format"),
+  BAD_RECORDS_LOGGER_ENABLE("bad_records_logger_enable"),
+  BAD_RECORDS_ACTION("bad_records_action");
+
+  private String name;
+
+  /**
+   * constructor to initialize the enum value
+   * @param name
+   */
+  TableOptionConstant(String name) {
+    this.name = name;
+  }
+
+  public String getName() {
+    return name;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/809a4d00/processing/src/main/java/org/apache/carbondata/processing/graphgenerator/GraphGenerator.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/graphgenerator/GraphGenerator.java
b/processing/src/main/java/org/apache/carbondata/processing/graphgenerator/GraphGenerator.java
index 1612ca1..1b5b68f 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/graphgenerator/GraphGenerator.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/graphgenerator/GraphGenerator.java
@@ -921,7 +921,7 @@ public class GraphGenerator {
     TableOptionWrapper tableOptionWrapper = TableOptionWrapper.getTableOptionWrapperInstance();
     tableOptionWrapper.setTableOption(schemaInfo.getSerializationNullFormat());
     tableOptionWrapper.setTableOption(schemaInfo.getBadRecordsLoggerEnable());
-    tableOptionWrapper.setTableOption(schemaInfo.getBadRecordsLoggerRedirect());
+    tableOptionWrapper.setTableOption(schemaInfo.getBadRecordsLoggerAction());
     return tableOptionWrapper;
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/809a4d00/processing/src/main/java/org/apache/carbondata/processing/surrogatekeysgenerator/csvbased/BadRecordslogger.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/surrogatekeysgenerator/csvbased/BadRecordslogger.java
b/processing/src/main/java/org/apache/carbondata/processing/surrogatekeysgenerator/csvbased/BadRecordslogger.java
index c373d62..ba33212 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/surrogatekeysgenerator/csvbased/BadRecordslogger.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/surrogatekeysgenerator/csvbased/BadRecordslogger.java
@@ -32,7 +32,6 @@ import java.util.Map;
 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.datastorage.store.filesystem.CarbonFile;
 import org.apache.carbondata.core.datastorage.store.impl.FileFactory;
 import org.apache.carbondata.core.datastorage.store.impl.FileFactory.FileType;
 import org.apache.carbondata.core.util.CarbonUtil;
@@ -69,9 +68,13 @@ public class BadRecordslogger {
   private BufferedWriter bufferedCSVWriter;
   private DataOutputStream outCSVStream;
   /**
-   *
+   * bad record log file path
+   */
+  private String logFilePath;
+  /**
+   * csv file path
    */
-  private CarbonFile logFile;
+  private String csvFilePath;
 
   /**
    * task key which is DatabaseName/TableName/tablename
@@ -145,14 +148,11 @@ public class BadRecordslogger {
    *
    */
   private synchronized void writeBadRecordsToFile(StringBuilder logStrings) {
-
-    if (null == logFile) {
-      String filePath =
+    if (null == logFilePath) {
+      logFilePath =
           this.storePath + File.separator + this.fileName + CarbonCommonConstants.LOG_FILE_EXTENSION
               + CarbonCommonConstants.FILE_INPROGRESS_STATUS;
-      logFile = FileFactory.getCarbonFile(filePath, FileFactory.getFileType(filePath));
     }
-
     try {
       if (null == bufferedWriter) {
         FileType fileType = FileFactory.getFileType(storePath);
@@ -161,13 +161,13 @@ public class BadRecordslogger {
           FileFactory.mkdirs(this.storePath, fileType);
 
           // create the files
-          FileFactory.createNewFile(logFile.getPath(), fileType);
+          FileFactory.createNewFile(logFilePath, fileType);
         }
 
-        outStream = FileFactory.getDataOutputStream(logFile.getPath(), fileType);
+        outStream = FileFactory.getDataOutputStream(logFilePath, fileType);
 
         bufferedWriter = new BufferedWriter(new OutputStreamWriter(outStream,
-                Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
+            Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
 
       }
       bufferedWriter.write(logStrings.toString());
@@ -185,12 +185,16 @@ public class BadRecordslogger {
   }
 
   /**
+   * method will write the row having bad record in the csv file.
    *
+   * @param logStrings
    */
   private synchronized void writeBadRecordsToCSVFile(StringBuilder logStrings) {
-    String filePath =
-        this.storePath + File.separator + this.fileName + CarbonCommonConstants.CSV_FILE_EXTENSION
-            + CarbonCommonConstants.FILE_INPROGRESS_STATUS;
+    if (null == csvFilePath) {
+      csvFilePath =
+          this.storePath + File.separator + this.fileName + CarbonCommonConstants.CSV_FILE_EXTENSION
+              + CarbonCommonConstants.FILE_INPROGRESS_STATUS;
+    }
     try {
       if (null == bufferedCSVWriter) {
         FileType fileType = FileFactory.getFileType(storePath);
@@ -199,10 +203,10 @@ public class BadRecordslogger {
           FileFactory.mkdirs(this.storePath, fileType);
 
           // create the files
-          FileFactory.createNewFile(filePath, fileType);
+          FileFactory.createNewFile(csvFilePath, fileType);
         }
 
-        outCSVStream = FileFactory.getDataOutputStream(filePath, fileType);
+        outCSVStream = FileFactory.getDataOutputStream(csvFilePath, fileType);
 
         bufferedCSVWriter = new BufferedWriter(new OutputStreamWriter(outCSVStream,
             Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
@@ -211,9 +215,9 @@ public class BadRecordslogger {
       bufferedCSVWriter.write(logStrings.toString());
       bufferedCSVWriter.newLine();
     } catch (FileNotFoundException e) {
-      LOGGER.error("Bad Log Files not found");
+      LOGGER.error("Bad record csv Files not found");
     } catch (IOException e) {
-      LOGGER.error("Error While writing bad log File");
+      LOGGER.error("Error While writing bad record csv File");
     }
     finally {
       badRecordEntry.put(taskKey, "Partially");

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/809a4d00/processing/src/main/java/org/apache/carbondata/processing/surrogatekeysgenerator/csvbased/CarbonCSVBasedSeqGenStep.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/surrogatekeysgenerator/csvbased/CarbonCSVBasedSeqGenStep.java
b/processing/src/main/java/org/apache/carbondata/processing/surrogatekeysgenerator/csvbased/CarbonCSVBasedSeqGenStep.java
index 8959179..94b2df8 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/surrogatekeysgenerator/csvbased/CarbonCSVBasedSeqGenStep.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/surrogatekeysgenerator/csvbased/CarbonCSVBasedSeqGenStep.java
@@ -75,6 +75,10 @@ import org.apache.carbondata.processing.schema.metadata.HierarchiesInfo;
 import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
 import org.apache.carbondata.processing.util.RemoveDictionaryUtil;
 
+import static org.apache.carbondata.processing.constants.TableOptionConstant.BAD_RECORDS_ACTION;
+import static org.apache.carbondata.processing.constants.TableOptionConstant.BAD_RECORDS_LOGGER_ENABLE;
+import static org.apache.carbondata.processing.constants.TableOptionConstant.SERIALIZATION_NULL_FORMAT;
+
 import org.pentaho.di.core.exception.KettleException;
 import org.pentaho.di.core.row.RowMetaInterface;
 import org.pentaho.di.core.row.ValueMeta;
@@ -439,12 +443,12 @@ public class CarbonCSVBasedSeqGenStep extends BaseStep {
           data.getSurrogateKeyGen()
               .setDimensionOrdinalToDimensionMapping(populateNameToCarbonDimensionMap());
         }
-        serializationNullFormat = meta.getTableOptionWrapper().get("serialization_null_format");
-        badRecordsLoggerEnable =
-            Boolean.parseBoolean(meta.getTableOptionWrapper().get("bad_records_logger_enable"));
-        badRecordConvertNullDisable = true;
+        serializationNullFormat =
+            meta.getTableOptionWrapper().get(SERIALIZATION_NULL_FORMAT.getName());
+        badRecordsLoggerEnable = Boolean
+            .parseBoolean(meta.getTableOptionWrapper().get(BAD_RECORDS_LOGGER_ENABLE.getName()));
         String bad_records_action =
-            meta.getTableOptionWrapper().get("bad_records_action");
+            meta.getTableOptionWrapper().get(BAD_RECORDS_ACTION.getName());
         if(null != bad_records_action) {
           LoggerAction loggerAction = null;
           try {
@@ -458,9 +462,11 @@ public class CarbonCSVBasedSeqGenStep extends BaseStep {
               break;
             case REDIRECT:
               badRecordsLogRedirect = true;
+              badRecordConvertNullDisable = true;
               break;
             case IGNORE:
               badRecordsLogRedirect = false;
+              badRecordConvertNullDisable = true;
               break;
           }
         }


Mime
View raw message