diff --git a/pom.xml b/pom.xml
index 2f69d97..08f27b8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -58,11 +58,16 @@
4.13.1
test
-
+
- com.googlecode.json-simple
- json-simple
- 1.1.1
+ com.fasterxml.jackson.core
+ jackson-core
+ 2.15.2
+
+
+ com.fasterxml.jackson.core
+ jackson-databind
+ 2.15.2
diff --git a/src/main/java/com/aerospike/load/AsWriterTask.java b/src/main/java/com/aerospike/load/AsWriterTask.java
index f4353c4..a34d8de 100644
--- a/src/main/java/com/aerospike/load/AsWriterTask.java
+++ b/src/main/java/com/aerospike/load/AsWriterTask.java
@@ -20,594 +20,601 @@
* IN THE SOFTWARE.
******************************************************************************/
-package com.aerospike.load;
-
-import java.text.DateFormat;
-import java.text.SimpleDateFormat;
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.List;
-import java.util.TreeMap;
-import java.util.concurrent.Callable;
-
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-import org.json.simple.JSONArray;
-import org.json.simple.JSONObject;
-import org.json.simple.parser.JSONParser;
-import org.json.simple.parser.ParseException;
-
-import com.aerospike.client.AerospikeClient;
-import com.aerospike.client.AerospikeException;
-import com.aerospike.client.Bin;
-import com.aerospike.client.Key;
-import com.aerospike.client.Value;
-import com.aerospike.client.ResultCode;
-
-/**
- *
- * @author Aerospike
- *
- * Main writer class to write data in Aerospike.
- *
- */
-public class AsWriterTask implements Callable {
-
- // File and line info variable.
- private String fileName;
- private int lineNumber;
- private int lineSize;
-
- // Aerospike related variable.
- private AerospikeClient client;
-
- // Data definition related variable
- private HashMap dsvConfigs;
- private MappingDefinition mappingDef;
- private List columns;
-
- private Parameters params;
- private Counter counters;
- private JSONParser jsonParser;
-
- private static Logger log = LogManager.getLogger(AsWriterTask.class);
-
- /**
- * AsWriterTask process given data columns for a record and create Set and Key and Bins.
- * It writes these Bins to created Key. If its secondary mapping then it will do CDT append
- * over all created Bins.
- *
- * @param fileName Name of the data file
- * @param lineNumber Line number in the file fileName
- * @param lineSize Size of the line to keep track of record processed
- * @param client AerospikeClient object
- * @param columns List of column separated entries in this lineNumber
- * @param dsvConfig Map of DSV configurations
- * @param mappingDef MappingDefinition of a mapping from config file
- * @param params User given parameters
- * @param counters Counter for stats
- *
- */
- public AsWriterTask(String fileName, int lineNumber, int lineSize,AerospikeClient client, List columns,
- HashMap dsvConfigs, MappingDefinition mappingDef, Parameters params, Counter counters) {
-
- // Passed to print log error with filename, line number, increment byteprocessed.
- this.fileName = fileName;
- this.lineNumber = lineNumber;
- this.lineSize = lineSize;
-
- this.client = client;
-
- this.dsvConfigs = dsvConfigs;
- this.mappingDef = mappingDef;
- this.columns = columns;
-
- this.params = params;
- this.counters = counters;
-
- }
-
- /*
- * Writes a record to the Aerospike Cluster
- */
- private void writeToAs(Key key, List bins) {
-
- try {
- // Connection could be broken at actual write time.
- if (this.client == null) {
- throw new Exception("Null Aerospike client !!");
- }
-
-
- if (bins.isEmpty()) {
- counters.write.noBinsSkipped.getAndIncrement();
- log.trace("No bins to insert");
- return;
- }
- // All bins will have append operation if secondary mapping.
- if (this.mappingDef.secondaryMapping) {
- for (Bin b : bins) {
- client.operate(this.params.writePolicy, key,
- com.aerospike.client.cdt.ListOperation.append(b.name, b.value));
- }
- counters.write.mappingWriteCount.getAndIncrement();
- } else {
- this.client.put(this.params.writePolicy, key, bins.toArray(new Bin[bins.size()]));
- counters.write.bytesProcessed.addAndGet(this.lineSize);
- }
- counters.write.writeCount.getAndIncrement();
-
- log.trace("Wrote line " + lineNumber + " Key: " + key.userKey + " to Aerospike.");
-
- } catch (AerospikeException ae) {
-
- handleAerospikeWriteError(ae);
- checkAndAbort();
-
- } catch (Exception e) {
-
- handleWriteError(e);
- checkAndAbort();
-
- }
- }
-
- private void handleAerospikeWriteError(AerospikeException ae) {
-
- log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber + "Aerospike Write Error: "
- + ae.getResultCode());
-
- if (log.isDebugEnabled()) {
- ae.printStackTrace();
- }
-
- switch (ae.getResultCode()) {
-
- case ResultCode.TIMEOUT:
- counters.write.writeTimeouts.getAndIncrement();
- break;
- case ResultCode.KEY_EXISTS_ERROR:
- counters.write.writeKeyExists.getAndIncrement();
- break;
- default:
- //..
- }
-
- if (!this.mappingDef.secondaryMapping) {
- counters.write.bytesProcessed.addAndGet(this.lineSize);
- }
-
- counters.write.writeErrors.getAndIncrement();
- }
-
- private void handleWriteError(Exception e) {
- log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber + " Write Error: " + e);
- if (log.isDebugEnabled()) {
- e.printStackTrace();
- }
- if (!this.mappingDef.secondaryMapping) {
- counters.write.bytesProcessed.addAndGet(this.lineSize);
- }
- counters.write.writeErrors.getAndIncrement();
- }
-
- private void checkAndAbort(){
- long errorTotal;
- errorTotal = (counters.write.readErrors.get() + counters.write.writeErrors.get()
- + counters.write.processingErrors.get());
- if (this.params.abortErrorCount != 0 && this.params.abortErrorCount < errorTotal) {
- System.exit(-1);
- }
- }
-
- /*
- * Create Set and Key from provided data for given mappingDef.
- * Create Bin for given binList in mappingDef.
- */
- private Key getKeyAndBinsFromDataline(List bins) {
- log.debug("processing File: " + Utils.getFileName(fileName) + "line: " + this.lineNumber);
- Key key = null;
-
+ package com.aerospike.load;
+
+ import java.text.DateFormat;
+ import java.text.SimpleDateFormat;
+ import java.util.ArrayList;
+ import java.util.Date;
+ import java.util.HashMap;
+ import java.util.List;
+ import java.util.TreeMap;
+ import java.util.concurrent.Callable;
+
+ import org.apache.logging.log4j.LogManager;
+ import org.apache.logging.log4j.Logger;
+ import java.io.IOException;
+ import java.util.Map;
+ import com.fasterxml.jackson.databind.ObjectMapper;
+ import com.fasterxml.jackson.core.JsonProcessingException;
+
+ import com.aerospike.client.AerospikeClient;
+ import com.aerospike.client.AerospikeException;
+ import com.aerospike.client.Bin;
+ import com.aerospike.client.Key;
+ import com.aerospike.client.Value;
+ import com.aerospike.client.ResultCode;
+
+ /**
+ *
+ * @author Aerospike
+ *
+ * Main writer class to write data in Aerospike.
+ *
+ */
+ public class AsWriterTask implements Callable {
+
+ // File and line info variable.
+ private String fileName;
+ private int lineNumber;
+ private int lineSize;
+
+ // Aerospike related variable.
+ private AerospikeClient client;
+
+ // Data definition related variable
+ private HashMap dsvConfigs;
+ private MappingDefinition mappingDef;
+ private List columns;
+
+ private Parameters params;
+ private Counter counters;
+ // JSON parsing is now handled by RelaxedJsonMapper
+
+ private static Logger log = LogManager.getLogger(AsWriterTask.class);
+
+ /**
+ * AsWriterTask process given data columns for a record and create Set and Key and Bins.
+ * It writes these Bins to created Key. If its secondary mapping then it will do CDT append
+ * over all created Bins.
+ *
+ * @param fileName Name of the data file
+ * @param lineNumber Line number in the file fileName
+ * @param lineSize Size of the line to keep track of record processed
+ * @param client AerospikeClient object
+ * @param columns List of column separated entries in this lineNumber
+ * @param dsvConfigs Map of DSV configurations
+ * @param mappingDef MappingDefinition of a mapping from config file
+ * @param params User given parameters
+ * @param counters Counter for stats
+ *
+ */
+ public AsWriterTask(String fileName, int lineNumber, int lineSize,AerospikeClient client, List columns,
+ HashMap dsvConfigs, MappingDefinition mappingDef, Parameters params, Counter counters) {
+
+ // Passed to print log error with filename, line number, increment byteprocessed.
+ this.fileName = fileName;
+ this.lineNumber = lineNumber;
+ this.lineSize = lineSize;
+
+ this.client = client;
+
+ this.dsvConfigs = dsvConfigs;
+ this.mappingDef = mappingDef;
+ this.columns = columns;
+
+ this.params = params;
+ this.counters = counters;
+
+ }
+
+ /*
+ * Writes a record to the Aerospike Cluster
+ */
+ private void writeToAs(Key key, List bins) {
+
+ try {
+ // Connection could be broken at actual write time.
+ if (this.client == null) {
+ throw new Exception("Null Aerospike client !!");
+ }
+
+
+ if (bins.isEmpty()) {
+ counters.write.noBinsSkipped.getAndIncrement();
+ log.trace("No bins to insert");
+ return;
+ }
+ // All bins will have append operation if secondary mapping.
+ if (this.mappingDef.secondaryMapping) {
+ for (Bin b : bins) {
+ client.operate(this.params.writePolicy, key,
+ com.aerospike.client.cdt.ListOperation.append(b.name, b.value));
+ }
+ counters.write.mappingWriteCount.getAndIncrement();
+ } else {
+ this.client.put(this.params.writePolicy, key, bins.toArray(new Bin[bins.size()]));
+ counters.write.bytesProcessed.addAndGet(this.lineSize);
+ }
+ counters.write.writeCount.getAndIncrement();
+
+ log.trace("Wrote line " + lineNumber + " Key: " + key.userKey + " to Aerospike.");
+
+ } catch (AerospikeException ae) {
+
+ handleAerospikeWriteError(ae);
+ checkAndAbort();
+
+ } catch (Exception e) {
+
+ handleWriteError(e);
+ checkAndAbort();
+
+ }
+ }
+
+ private void handleAerospikeWriteError(AerospikeException ae) {
+
+ log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber + "Aerospike Write Error: "
+ + ae.getResultCode());
+
+ if (log.isDebugEnabled()) {
+ ae.printStackTrace();
+ }
+
+ switch (ae.getResultCode()) {
+
+ case ResultCode.TIMEOUT:
+ counters.write.writeTimeouts.getAndIncrement();
+ break;
+ case ResultCode.KEY_EXISTS_ERROR:
+ counters.write.writeKeyExists.getAndIncrement();
+ break;
+ default:
+ //..
+ }
+
+ if (!this.mappingDef.secondaryMapping) {
+ counters.write.bytesProcessed.addAndGet(this.lineSize);
+ }
+
+ counters.write.writeErrors.getAndIncrement();
+ }
+
+ private void handleWriteError(Exception e) {
+ log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber + " Write Error: " + e);
+ if (log.isDebugEnabled()) {
+ e.printStackTrace();
+ }
+ if (!this.mappingDef.secondaryMapping) {
+ counters.write.bytesProcessed.addAndGet(this.lineSize);
+ }
+ counters.write.writeErrors.getAndIncrement();
+ }
+
+ private void checkAndAbort(){
+ long errorTotal;
+ errorTotal = (counters.write.readErrors.get() + counters.write.writeErrors.get()
+ + counters.write.processingErrors.get());
+ if (this.params.abortErrorCount != 0 && this.params.abortErrorCount < errorTotal) {
+ System.exit(-1);
+ }
+ }
+
+ /*
+ * Create Set and Key from provided data for given mappingDef.
+ * Create Bin for given binList in mappingDef.
+ */
+ private Key getKeyAndBinsFromDataline(List bins) {
+ log.debug("processing File: " + Utils.getFileName(fileName) + "line: " + this.lineNumber);
+ Key key = null;
+
+ try {
+
+ validateNColumnInDataline();
+
+ // Set couldn't be null here. Its been validated earlier.
+ String set = getSetName();
+
+ key = createRecordKey(this.params.namespace, set);
+
+ populateAsBinFromColumnDef(bins);
+
+ log.trace("Formed key and bins for line: " + lineNumber + " Key: " + key.userKey + " Bins: "
+ + bins.toString());
+
+ } catch (AerospikeException ae) {
+
+ log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber
+ + " Aerospike Bin processing Error: " + ae.getResultCode());
+ handleProcessLineError(ae);
+
+ } catch (Exception e) {
+
+ log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber + " Error: " + e);
+ handleProcessLineError(e);
+
+ }
+ return key;
+ }
+
+ /*
+ * Validate if number of column in data line are same as provided in config file.
+ * Throw exception the more columns are present then given.
+ */
+ private void validateNColumnInDataline() throws Exception {
+
+ // Throw exception if n_columns(datafile) are more than n_columns(configfile).
+ int n_column = Integer.parseInt(dsvConfigs.get(Constants.N_COLUMN));
+ if (columns.size() == n_column) {
+ return;
+ }
+ if (columns.size() < n_column) {
+ log.warn("File: " + Utils.getFileName(fileName) + " Line: " + lineNumber
+ + " Number of column mismatch:Columns in data file is less than number of column in config file.");
+ } else {
+ throw new Exception("Column count mismatch at line " + lineNumber);
+ }
+ }
+
+ private void handleProcessLineError(Exception e) {
+ if (log.isDebugEnabled()) {
+ e.printStackTrace();
+ }
+ counters.write.processingErrors.getAndIncrement();
+ counters.write.bytesProcessed.addAndGet(this.lineSize);
+ checkAndAbort();
+ }
+
+
+ private String getSetName() {
+ MetaDefinition setColumn = this.mappingDef.setColumnDef;
+
+ if (setColumn.staticName != null) {
+ return setColumn.staticName;
+ }
+
+ String set = null;
+ String setRawText = this.columns.get(setColumn.nameDef.columnPos);
+ if (setColumn.nameDef.removePrefix != null) {
+ if (setRawText != null && setRawText.startsWith(setColumn.nameDef.removePrefix)) {
+ set = setRawText.substring(setColumn.nameDef.removePrefix.length());
+ }
+ } else {
+ set = setRawText;
+ }
+ return set;
+ }
+
+ private Key createRecordKey(String namespace, String set) throws Exception {
+ // Use 'SET' name to create key.
+ Key key = null;
+
+ MetaDefinition keyColumn = this.mappingDef.keyColumnDef;
+
+ String keyRawText = this.columns.get(keyColumn.nameDef.columnPos);
+
+ if (keyRawText == null || keyRawText.trim().length() == 0) {
+ counters.write.keyNullSkipped.getAndIncrement();
+ throw new Exception("Key value is null in datafile.");
+ }
+
+ if ((keyColumn.nameDef.removePrefix != null)
+ && (keyRawText.startsWith(keyColumn.nameDef.removePrefix))) {
+ keyRawText = keyRawText.substring(keyColumn.nameDef.removePrefix.length());
+ }
+
+ if (keyColumn.nameDef.srcType == SrcColumnType.INTEGER) {
+ Long integer = Long.parseLong(keyRawText);
+ key = new Key(namespace, set, integer);
+ } else {
+ key = new Key(namespace, set, keyRawText);
+ }
+
+ return key;
+ }
+
+ private void populateAsBinFromColumnDef(List bins) {
+ for (BinDefinition binColumn : this.mappingDef.binColumnDefs) {
+ Bin bin = null;
+ String binName = null;
+ String binRawValue = null;
+
+ // Get binName.
+ if (binColumn.staticName != null) {
+ binName = binColumn.staticName;
+ } else if (binColumn.nameDef != null) {
+ binName = this.columns.get(binColumn.nameDef.columnPos);
+ }
+
+ // Get BinValue.
+ if (binColumn.staticValue != null) {
+
+ binRawValue = binColumn.staticValue;
+ bin = new Bin(binName, binRawValue);
+
+ } else if (binColumn.valueDef != null) {
+
+ binRawValue = getbinRawValue(binColumn);
+ if (binRawValue == null || binRawValue.equals("")) {
+ continue;
+ }
+
+ switch (binColumn.valueDef.srcType) {
+
+ case INTEGER:
+ bin = createBinForInteger(binName, binRawValue);
+ break;
+ case FLOAT:
+ bin = createBinForFloat(binName, binRawValue);
+ break;
+ case STRING:
+ bin = createBinForString(binName, binRawValue);
+ break;
+ case JSON:
+ /*
+ * JSON could take any valid JSON. There are two type of JSON:
+ * JsonArray: this can be used to insert List (Any generic JSON list)
+ * JsonObj: this can be used to insert Map (Any generic JSON object)
+ */
+ bin = createBinForJson(binName, binRawValue);
+ break;
+ case GEOJSON:
+ bin = createBinForGeoJson(binName, binRawValue);
+ break;
+ case BLOB:
+ bin = createBinForBlob(binColumn, binName, binRawValue);
+ break;
+ case TIMESTAMP:
+ bin = createBinForTimestamp(binColumn, binName, binRawValue);
+ break;
+ default:
+ //....
+ }
+ }
+
+ if (bin != null) {
+ bins.add(bin);
+ }
+ }
+ }
+
+ private String getbinRawValue(BinDefinition binColumn) {
+ /*
+ * User may want to store the time when record is written in Aerospike.
+ * Assign system_time to binvalue. This bin will be written as part of
+ * record.
+ */
+ if (binColumn.valueDef.columnName != null
+ && binColumn.valueDef.columnName.toLowerCase().equals(Constants.SYSTEM_TIME)) {
+
+ SimpleDateFormat sdf =
+ new SimpleDateFormat(binColumn.valueDef.encoding); // dd/MM/yyyy
+ Date now = new Date();
+ return sdf.format(now);
+ }
+
+
+ String binRawValue = this.columns.get(binColumn.valueDef.columnPos);
+
+ if ((binColumn.valueDef.removePrefix != null)
+ && (binRawValue != null && binRawValue.startsWith(binColumn.valueDef.removePrefix))) {
+ binRawValue =
+ binRawValue.substring(binColumn.valueDef.removePrefix.length());
+ }
+ return binRawValue;
+ }
+
+ private Bin createBinForInteger(String binName, String binRawValue) {
+
+ try {
+ // Server stores all integer type data in 64bit so use long.
+ Long integer = Long.parseLong(binRawValue);
+
+ return new Bin(binName, integer);
+
+ } catch (Exception pi) {
+
+ log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber
+ + " Integer/Long Parse Error: " + pi);
+ return null;
+
+ }
+ }
+
+ private Bin createBinForFloat(String binName, String binRawValue) {
+
+ try {
+ // parse as a double to get greater precision
+ double binfloat = Double.parseDouble(binRawValue);
+
+ return new Bin(binName, binfloat);
+
+ } catch (Exception e) {
+ log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber
+ + " Floating number Parse Error: " + e);
+ return null;
+ }
+
+ }
+
+ private Bin createBinForString(String binName, String binRawValue) {
+ return new Bin(binName, binRawValue);
+ }
+
+ private Bin createBinForJson(String binName, String binRawValue) {
try {
-
- validateNColumnInDataline();
-
- // Set couldn't be null here. Its been validated earlier.
- String set = getSetName();
-
- key = createRecordKey(this.params.namespace, set);
+ log.debug(binRawValue);
- populateAsBinFromColumnDef(bins);
+ Object parsedObject = null;
- log.trace("Formed key and bins for line: " + lineNumber + " Key: " + key.userKey + " Bins: "
- + bins.toString());
-
- } catch (AerospikeException ae) {
-
- log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber
- + " Aerospike Bin processing Error: " + ae.getResultCode());
- handleProcessLineError(ae);
-
- } catch (ParseException pe) {
-
- log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber + " Parsing Error: " + pe);
- handleProcessLineError(pe);
-
- } catch (Exception e) {
-
- log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber + " Unknown Error: " + e);
- handleProcessLineError(e);
-
- }
- return key;
- }
-
- /*
- * Validate if number of column in data line are same as provided in config file.
- * Throw exception the more columns are present then given.
- */
- private void validateNColumnInDataline() throws ParseException {
-
- // Throw exception if n_columns(datafile) are more than n_columns(configfile).
- int n_column = Integer.parseInt(dsvConfigs.get(Constants.N_COLUMN));
- if (columns.size() == n_column) {
- return;
- }
- if (columns.size() < n_column) {
- log.warn("File: " + Utils.getFileName(fileName) + " Line: " + lineNumber
- + " Number of column mismatch:Columns in data file is less than number of column in config file.");
- } else {
- throw new ParseException(lineNumber);
- }
- }
-
- private void handleProcessLineError(Exception e) {
- if (log.isDebugEnabled()) {
- e.printStackTrace();
- }
- counters.write.processingErrors.getAndIncrement();
- counters.write.bytesProcessed.addAndGet(this.lineSize);
- checkAndAbort();
- }
-
-
- private String getSetName() {
- MetaDefinition setColumn = this.mappingDef.setColumnDef;
-
- if (setColumn.staticName != null) {
- return setColumn.staticName;
- }
-
- String set = null;
- String setRawText = this.columns.get(setColumn.nameDef.columnPos);
- if (setColumn.nameDef.removePrefix != null) {
- if (setRawText != null && setRawText.startsWith(setColumn.nameDef.removePrefix)) {
- set = setRawText.substring(setColumn.nameDef.removePrefix.length());
- }
- } else {
- set = setRawText;
- }
- return set;
- }
-
- private Key createRecordKey(String namespace, String set) throws Exception {
- // Use 'SET' name to create key.
- Key key = null;
-
- MetaDefinition keyColumn = this.mappingDef.keyColumnDef;
-
- String keyRawText = this.columns.get(keyColumn.nameDef.columnPos);
-
- if (keyRawText == null || keyRawText.trim().length() == 0) {
- counters.write.keyNullSkipped.getAndIncrement();
- throw new Exception("Key value is null in datafile.");
- }
-
- if ((keyColumn.nameDef.removePrefix != null)
- && (keyRawText.startsWith(keyColumn.nameDef.removePrefix))) {
- keyRawText = keyRawText.substring(keyColumn.nameDef.removePrefix.length());
- }
-
- if (keyColumn.nameDef.srcType == SrcColumnType.INTEGER) {
- Long integer = Long.parseLong(keyRawText);
- key = new Key(namespace, set, integer);
- } else {
- key = new Key(namespace, set, keyRawText);
- }
-
- return key;
- }
-
- private void populateAsBinFromColumnDef(List bins) {
- for (BinDefinition binColumn : this.mappingDef.binColumnDefs) {
- Bin bin = null;
- String binName = null;
- String binRawValue = null;
-
- // Get binName.
- if (binColumn.staticName != null) {
- binName = binColumn.staticName;
- } else if (binColumn.nameDef != null) {
- binName = this.columns.get(binColumn.nameDef.columnPos);
- }
-
- // Get BinValue.
- if (binColumn.staticValue != null) {
-
- binRawValue = binColumn.staticValue;
- bin = new Bin(binName, binRawValue);
-
- } else if (binColumn.valueDef != null) {
-
- binRawValue = getbinRawValue(binColumn);
- if (binRawValue == null || binRawValue.equals("")) {
- continue;
- }
-
- switch (binColumn.valueDef.srcType) {
-
- case INTEGER:
- bin = createBinForInteger(binName, binRawValue);
- break;
- case FLOAT:
- bin = createBinForFloat(binName, binRawValue);
- break;
- case STRING:
- bin = createBinForString(binName, binRawValue);
- break;
- case JSON:
- /*
- * JSON could take any valid JSON. There are two type of JSON:
- * JsonArray: this can be used to insert List (Any generic JSON list)
- * JsonObj: this can be used to insert Map (Any generic JSON object)
- */
- bin = createBinForJson(binName, binRawValue);
- break;
- case GEOJSON:
- bin = createBinForGeoJson(binName, binRawValue);
- break;
- case BLOB:
- bin = createBinForBlob(binColumn, binName, binRawValue);
- break;
- case TIMESTAMP:
- bin = createBinForTimestamp(binColumn, binName, binRawValue);
- break;
- default:
- //....
- }
- }
-
- if (bin != null) {
- bins.add(bin);
- }
- }
- }
-
- private String getbinRawValue(BinDefinition binColumn) {
- /*
- * User may want to store the time when record is written in Aerospike.
- * Assign system_time to binvalue. This bin will be written as part of
- * record.
- */
- if (binColumn.valueDef.columnName != null
- && binColumn.valueDef.columnName.toLowerCase().equals(Constants.SYSTEM_TIME)) {
-
- SimpleDateFormat sdf =
- new SimpleDateFormat(binColumn.valueDef.encoding); // dd/MM/yyyy
- Date now = new Date();
- return sdf.format(now);
- }
-
-
- String binRawValue = this.columns.get(binColumn.valueDef.columnPos);
-
- if ((binColumn.valueDef.removePrefix != null)
- && (binRawValue != null && binRawValue.startsWith(binColumn.valueDef.removePrefix))) {
- binRawValue =
- binRawValue.substring(binColumn.valueDef.removePrefix.length());
- }
- return binRawValue;
- }
-
- private Bin createBinForInteger(String binName, String binRawValue) {
-
- try {
- // Server stores all integer type data in 64bit so use long.
- Long integer = Long.parseLong(binRawValue);
-
- return new Bin(binName, integer);
-
- } catch (Exception pi) {
-
- log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber
- + " Integer/Long Parse Error: " + pi);
- return null;
-
- }
- }
-
- private Bin createBinForFloat(String binName, String binRawValue) {
-
- try {
- // parse as a double to get greater precision
- double binfloat = Double.parseDouble(binRawValue);
-
- return new Bin(binName, binfloat);
-
- } catch (Exception e) {
- log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber
- + " Floating number Parse Error: " + e);
- return null;
- }
-
- }
-
- private Bin createBinForString(String binName, String binRawValue) {
- return new Bin(binName, binRawValue);
- }
-
- private Bin createBinForJson(String binName, String binRawValue) {
- try {
- log.debug(binRawValue);
-
- if (jsonParser == null) {
- jsonParser = new JSONParser();
+ try {
+ ObjectMapper standardMapper = new ObjectMapper();
+ parsedObject = standardMapper.readValue(binRawValue, Object.class);
+ log.debug("Successfully parsed with standard JSON parser");
+ } catch (JsonProcessingException standardParseException) {
+ log.debug("Standard JSON parsing failed, using relaxed parser: " + standardParseException.getMessage());
+ parsedObject = RelaxedJsonMapper.parseJsonWithTypeHandling(binRawValue);
}
-
- Object obj = jsonParser.parse(binRawValue);
-
- if (obj instanceof JSONArray) {
- JSONArray jsonArray = (JSONArray) obj;
+
+ // Apply the same logic regardless of which parser was used
+ if (parsedObject instanceof List) {
+ List jsonArray = (List) parsedObject;
return new Bin(binName, jsonArray);
- } else {
- JSONObject jsonObj = (JSONObject) obj;
+ } else if (parsedObject instanceof Map) {
+ Map jsonObj = (Map) parsedObject;
if (this.params.unorderdMaps) {
- return new Bin(binName, jsonObj);
+ return new Bin(binName, jsonObj);
}
- TreeMap sortedMap = new TreeMap<>();
- sortedMap.putAll(jsonObj);
- return new Bin(binName, sortedMap);
+ try {
+ TreeMap sortedMap = new TreeMap<>();
+ sortedMap.putAll(jsonObj);
+ return new Bin(binName, sortedMap);
+ } catch (ClassCastException e) {
+ log.debug("TreeMap failed due to non-comparable keys, using unordered map: " + e.getMessage());
+ return new Bin(binName, jsonObj);
+ }
+ } else {
+ return new Bin(binName, parsedObject.toString());
}
-
- } catch (ParseException e) {
+
+ } catch (IOException e) {
log.error("Failed to parse JSON: " + e);
return null;
}
-
- }
-
- private Bin createBinForGeoJson(String binName, String binRawValue) {
- try {
- return new Bin(binName, Value.getAsGeoJSON(binRawValue));
- } catch (Exception e) {
- log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber
- + " GeoJson Parse Error: " + e);
- return null;
- }
- }
-
- private Bin createBinForBlob(BinDefinition binColumn, String binName, String binRawValue) {
- try {
- if ((binColumn.valueDef.dstType.equals(DstColumnType.BLOB))
- && (binColumn.valueDef.encoding.equalsIgnoreCase(Constants.HEX_ENCODING))) {
- return new Bin(binName, this.toByteArray(binRawValue));
- }
- } catch (Exception e) {
- log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber
- + " Blob Parse Error: " + e);
- return null;
- }
-
- return null;
- }
-
- private Bin createBinForTimestamp(BinDefinition binColumn, String binName, String binRawValue) {
-
- if (! binColumn.valueDef.dstType.equals(DstColumnType.INTEGER)) {
- return new Bin(binName, binRawValue);
- }
-
- DateFormat format = new SimpleDateFormat(binColumn.valueDef.encoding);
-
- try {
-
- Date formatDate = format.parse(binRawValue);
- long milliSecondForDate = formatDate.getTime() - this.params.timeZoneOffset;
-
- if (!(binColumn.valueDef.encoding.contains(".SSS")
- && binName.equals(Constants.SYSTEM_TIME))) {
- // We need time in milliseconds so no need to change it to milliseconds.
- milliSecondForDate = milliSecondForDate / 1000;
- }
-
- log.trace("Date format: " + binRawValue + " in seconds: " + milliSecondForDate);
-
- return new Bin(binName, milliSecondForDate);
-
- } catch (java.text.ParseException e) {
- e.printStackTrace();
- return null;
- }
-
- }
-
- private byte[] toByteArray(String s) {
-
- if ((s.length() % 2) != 0) {
- log.error("blob exception: " + s);
- throw new IllegalArgumentException("Input hex formated string must contain an even number of characters.");
- }
-
- int len = s.length();
- byte[] data = new byte[len / 2];
-
- try {
- for (int i = 0; i < len; i += 2) {
- data[i / 2] = (byte) ((Character.digit(s.charAt(i), 16) << 4) + Character.digit(s.charAt(i + 1), 16));
- }
- } catch (Exception e) {
- log.error("blob exception: " + e);
- }
- return data;
- }
-
- private boolean exceedingThroughput() {
- long transactions;
- long timeLapse;
- long throughput;
-
- transactions = counters.write.writeCount.get()
- + counters.write.mappingWriteCount.get()
- + counters.write.writeErrors.get();
-
-
- timeLapse = (System.currentTimeMillis() - counters.write.writeStartTime) / 1000L;
-
- if (timeLapse > 0) {
- throughput = transactions / timeLapse;
-
- if (throughput > params.maxThroughput) {
- return true;
- }
- }
- return false;
- }
- public Integer call() throws Exception {
-
- List bins = new ArrayList();
-
-
- try {
-
- counters.write.processingCount.getAndIncrement();
-
- Key key = getKeyAndBinsFromDataline(bins);
-
- if (key != null) {
- writeToAs(key, bins);
- bins.clear();
-
- if (params.maxThroughput == 0) {
- return 0;
- }
-
- while(exceedingThroughput()) {
- Thread.sleep(20);
- }
-
- return 0;
- }
-
- } catch (Exception e) {
-
- log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber + " Parsing Error: " + e);
- log.debug(e);
- }
-
- return 0;
-
}
-}
+
+ private Bin createBinForGeoJson(String binName, String binRawValue) {
+ try {
+ return new Bin(binName, Value.getAsGeoJSON(binRawValue));
+ } catch (Exception e) {
+ log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber
+ + " GeoJson Parse Error: " + e);
+ return null;
+ }
+ }
+
+ private Bin createBinForBlob(BinDefinition binColumn, String binName, String binRawValue) {
+ try {
+ if ((binColumn.valueDef.dstType.equals(DstColumnType.BLOB))
+ && (binColumn.valueDef.encoding.equalsIgnoreCase(Constants.HEX_ENCODING))) {
+ return new Bin(binName, this.toByteArray(binRawValue));
+ }
+ } catch (Exception e) {
+ log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber
+ + " Blob Parse Error: " + e);
+ return null;
+ }
+
+ return null;
+ }
+
+ private Bin createBinForTimestamp(BinDefinition binColumn, String binName, String binRawValue) {
+
+ if (! binColumn.valueDef.dstType.equals(DstColumnType.INTEGER)) {
+ return new Bin(binName, binRawValue);
+ }
+
+ DateFormat format = new SimpleDateFormat(binColumn.valueDef.encoding);
+
+ try {
+
+ Date formatDate = format.parse(binRawValue);
+ long milliSecondForDate = formatDate.getTime() - this.params.timeZoneOffset;
+
+ if (!(binColumn.valueDef.encoding.contains(".SSS")
+ && binName.equals(Constants.SYSTEM_TIME))) {
+ // We need time in milliseconds so no need to change it to milliseconds.
+ milliSecondForDate = milliSecondForDate / 1000;
+ }
+
+ log.trace("Date format: " + binRawValue + " in seconds: " + milliSecondForDate);
+
+ return new Bin(binName, milliSecondForDate);
+
+ } catch (java.text.ParseException e) {
+ e.printStackTrace();
+ return null;
+ }
+
+ }
+
+ private byte[] toByteArray(String s) {
+
+ if ((s.length() % 2) != 0) {
+ log.error("blob exception: " + s);
+ throw new IllegalArgumentException("Input hex formated string must contain an even number of characters.");
+ }
+
+ int len = s.length();
+ byte[] data = new byte[len / 2];
+
+ try {
+ for (int i = 0; i < len; i += 2) {
+ data[i / 2] = (byte) ((Character.digit(s.charAt(i), 16) << 4) + Character.digit(s.charAt(i + 1), 16));
+ }
+ } catch (Exception e) {
+ log.error("blob exception: " + e);
+ }
+ return data;
+ }
+
+ private boolean exceedingThroughput() {
+ long transactions;
+ long timeLapse;
+ long throughput;
+
+ transactions = counters.write.writeCount.get()
+ + counters.write.mappingWriteCount.get()
+ + counters.write.writeErrors.get();
+
+
+ timeLapse = (System.currentTimeMillis() - counters.write.writeStartTime) / 1000L;
+
+ if (timeLapse > 0) {
+ throughput = transactions / timeLapse;
+
+ if (throughput > params.maxThroughput) {
+ return true;
+ }
+ }
+ return false;
+ }
+ public Integer call() throws Exception {
+
+ List bins = new ArrayList();
+
+
+ try {
+
+ counters.write.processingCount.getAndIncrement();
+
+ Key key = getKeyAndBinsFromDataline(bins);
+
+ if (key != null) {
+ writeToAs(key, bins);
+ bins.clear();
+
+ if (params.maxThroughput == 0) {
+ return 0;
+ }
+
+ while(exceedingThroughput()) {
+ Thread.sleep(20);
+ }
+
+ return 0;
+ }
+
+ } catch (Exception e) {
+
+ log.error("File: " + Utils.getFileName(this.fileName) + " Line: " + lineNumber + " Parsing Error: " + e);
+ log.debug(e);
+ }
+
+ return 0;
+
+ }
+ }
\ No newline at end of file
diff --git a/src/main/java/com/aerospike/load/Parser.java b/src/main/java/com/aerospike/load/Parser.java
index 7d5a6e5..6cec1d4 100644
--- a/src/main/java/com/aerospike/load/Parser.java
+++ b/src/main/java/com/aerospike/load/Parser.java
@@ -32,10 +32,8 @@
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
-import org.json.simple.JSONArray;
-import org.json.simple.JSONObject;
-import org.json.simple.parser.JSONParser;
-import org.json.simple.parser.ParseException;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
/**
* Parser class to parse different schema/(data definition) and data files.
@@ -51,28 +49,21 @@ public class Parser {
* @param configFile Config/schema/definition file name
* @param dsvConfigs DSV configuration for loader to use, given in config file (version, n_columns, delimiter...)
* @param mappingDefs List of schema/definitions for all mappings. primary + secondary mappings.
- * @param params User given parameters
- * @throws ParseException
* @throws IOException
*/
public static boolean parseJSONColumnDefinitions(File configFile, HashMap dsvConfigs, List mappingDefs) {
FileReader fr = null;
try {
- JSONParser jsonParser = new JSONParser();
-
- Object obj;
fr = new FileReader(configFile);
- obj = jsonParser.parse(fr);
+ JsonNode jobj = RelaxedJsonMapper.parseJson(fr);
- JSONObject jobj;
- if (obj == null) {
+ if (jobj == null) {
log.error("Empty config File.");
if (fr != null)
fr.close();
return false;
} else {
- jobj = (JSONObject) obj;
- log.debug("Config file contents: " + jobj.toJSONString());
+ log.debug("Config file contents: " + RelaxedJsonMapper.jsonNodeToString(jobj));
}
/*
@@ -91,12 +82,6 @@ public static boolean parseJSONColumnDefinitions(File configFile, HashMap dsvConfigs) {
+ private static boolean getUpdateDsvConfig(JsonNode jobj, HashMap dsvConfigs) {
Object obj = null;
// Get Metadata of loader (update dsvConfigs)
- if ((obj = getFromJsonObject(jobj, Constants.VERSION)) == null) {
+ if ((obj = RelaxedJsonMapper.getFromJsonNode(jobj, Constants.VERSION)) == null) {
log.error("\"" + Constants.VERSION + "\" Key is missing in config file.");
return false;
}
@@ -134,23 +119,23 @@ private static boolean getUpdateDsvConfig(JSONObject jobj, HashMap mappingDefs) throws Exception {
+ private static boolean getUpdateMappingColumnDefs(JsonNode jobj, List mappingDefs) throws Exception {
Object obj = null;
- JSONArray mappings;
- if ((obj = getFromJsonObject(jobj, Constants.MAPPINGS)) != null) {
- mappings = (JSONArray) obj;
- Iterator it = mappings.iterator();
- while (it.hasNext()) {
- JSONObject mappingObj = (JSONObject) it.next();
- MappingDefinition md = getMappingDef(mappingObj);
- if (md != null) {
- mappingDefs.add(md);
- } else {
- log.error("Error in parsing mappingdef: " + mappingObj.toString());
- return false;
+ JsonNode mappings;
+ if ((obj = RelaxedJsonMapper.getFromJsonNode(jobj, Constants.MAPPINGS)) != null) {
+ mappings = jobj.get(Constants.MAPPINGS);
+ if (mappings.isArray()) {
+ for (JsonNode mappingObj : mappings) {
+ MappingDefinition md = getMappingDef(mappingObj);
+ if (md != null) {
+ mappingDefs.add(md);
+ } else {
+ log.error("Error in parsing mappingdef: " + RelaxedJsonMapper.jsonNodeToString(mappingObj));
+ return false;
+ }
}
}
}
return true;
}
- private static Object getFromJsonObject(JSONObject jobj, String key) {
- return jobj.get(key);
+ private static Object getFromJsonObject(JsonNode jobj, String key) {
+ return RelaxedJsonMapper.getFromJsonNode(jobj, key);
}
/*
* Parse mapping definition from config file to get mappingDef
* This will have (secondary_mapping_ keyDefinition, setDefinition, binDefinition).
*/
- private static MappingDefinition getMappingDef(JSONObject mappingObj) throws Exception {
+ private static MappingDefinition getMappingDef(JsonNode mappingObj) throws Exception {
boolean secondary_mapping = false;
MetaDefinition keyColumnDef = null;
@@ -203,33 +188,33 @@ private static MappingDefinition getMappingDef(JSONObject mappingObj) throws Exc
}
if ((obj = getFromJsonObject(mappingObj, Constants.KEY)) != null) {
- keyColumnDef = getMetaDefs((JSONObject) obj, Constants.KEY);
+ keyColumnDef = getMetaDefs(mappingObj.get(Constants.KEY), Constants.KEY);
} else {
- log.error("\"" + Constants.KEY + "\" Key is missing in mapping. Mapping: " + mappingObj.toString());
+ log.error("\"" + Constants.KEY + "\" Key is missing in mapping. Mapping: " + RelaxedJsonMapper.jsonNodeToString(mappingObj));
return null;
}
if ((obj = getFromJsonObject(mappingObj, Constants.SET)) == null) {
- log.error("\"" + Constants.SET + "\" Key is missing in mapping. Mapping: " + mappingObj.toString());
+ log.error("\"" + Constants.SET + "\" Key is missing in mapping. Mapping: " + RelaxedJsonMapper.jsonNodeToString(mappingObj));
return null;
} else if (obj instanceof String) {
setColumnDef = new MetaDefinition(obj.toString(), null);
} else {
- setColumnDef = getMetaDefs((JSONObject) obj, Constants.SET);
+ setColumnDef = getMetaDefs(mappingObj.get(Constants.SET), Constants.SET);
}
if ((obj = getFromJsonObject(mappingObj, Constants.BINLIST)) != null) {
- JSONArray binObjList = (JSONArray) obj;
- Iterator it = binObjList.iterator();
- while (it.hasNext()) {
- JSONObject binObj = (JSONObject) it.next();
- BinDefinition binDef = getBinDefs(binObj);
- if (binDef != null) {
- binColumnDefs.add(binDef);
- } else {
- log.error("Error in parsing binDef: " + binObj.toString());
- return null;
+ JsonNode binObjList = mappingObj.get(Constants.BINLIST);
+ if (binObjList.isArray()) {
+ for (JsonNode binObj : binObjList) {
+ BinDefinition binDef = getBinDefs(binObj);
+ if (binDef != null) {
+ binColumnDefs.add(binDef);
+ } else {
+ log.error("Error in parsing binDef: " + RelaxedJsonMapper.jsonNodeToString(binObj));
+ return null;
+ }
}
}
} else {
@@ -243,7 +228,7 @@ private static MappingDefinition getMappingDef(JSONObject mappingObj) throws Exc
/*
* Parsing Meta definition(for Set or Key) from config file and populate metaDef object.
*/
- private static MetaDefinition getMetaDefs(JSONObject jobj, String jobjName) {
+ private static MetaDefinition getMetaDefs(JsonNode jobj, String jobjName) {
// Parsing Key, Set definition
ColumnDefinition valueDef = new ColumnDefinition(-1, null, null, null, null, null, null);
@@ -253,14 +238,15 @@ private static MetaDefinition getMetaDefs(JSONObject jobj, String jobjName) {
} else if ((jobj.get(Constants.COLUMN_NAME)) != null) {
- valueDef.columnName = (String) (jobj.get(Constants.COLUMN_NAME));
+ valueDef.columnName = jobj.get(Constants.COLUMN_NAME).asText();
} else {
log.error("Column_name or pos info is missing. Specify proper key/set mapping in config file for: " + jobjName + ":"
- + jobj.toString());
+ + RelaxedJsonMapper.jsonNodeToString(jobj));
}
- valueDef.setSrcType((String) jobj.get(Constants.TYPE));
+ JsonNode typeNode = jobj.get(Constants.TYPE);
+ valueDef.setSrcType(typeNode != null ? typeNode.asText() : null);
// Default set type is 'string'. what is default key type?
if (Constants.SET.equalsIgnoreCase(jobjName) && valueDef.srcType == null) {
@@ -268,8 +254,9 @@ private static MetaDefinition getMetaDefs(JSONObject jobj, String jobjName) {
}
// Get prefix to remove. Prefix will be removed from data
- if ((jobj.get(Constants.REMOVE_PREFIX)) != null) {
- valueDef.removePrefix = (String) jobj.get(Constants.REMOVE_PREFIX);
+ JsonNode prefixNode = jobj.get(Constants.REMOVE_PREFIX);
+ if (prefixNode != null) {
+ valueDef.removePrefix = prefixNode.asText();
}
return new MetaDefinition(null, valueDef);
@@ -278,7 +265,7 @@ private static MetaDefinition getMetaDefs(JSONObject jobj, String jobjName) {
/*
* Parsing Bin definition from config file and populate BinDef object
*/
- private static BinDefinition getBinDefs(JSONObject jobj) {
+ private static BinDefinition getBinDefs(JsonNode jobj) {
/*
* Sample Bin object
* {"name": "age", "value": {"column_name": "age", "type" : "integer"} }
@@ -290,23 +277,23 @@ private static BinDefinition getBinDefs(JSONObject jobj) {
ColumnDefinition nameDef = new ColumnDefinition(-1, null, null, null, null, null, null);
String staticBinName = null;
- if ((obj = jobj.get(Constants.NAME)) == null) {
- log.error(Constants.NAME + " key is missing object: " + jobj.toString());
+ JsonNode nameNode = jobj.get(Constants.NAME);
+ if (nameNode == null) {
+ log.error(Constants.NAME + " key is missing object: " + RelaxedJsonMapper.jsonNodeToString(jobj));
return null;
- } else if (!(obj instanceof JSONObject)) {
- staticBinName = (String) obj;
+ } else if (nameNode.isTextual()) {
+ staticBinName = nameNode.asText();
} else {
- JSONObject nameObj = (JSONObject) obj;
- if ((nameObj.get(Constants.COLUMN_POSITION)) != null) {
-
- nameDef.columnPos = (Integer.parseInt(nameObj.get(Constants.COLUMN_POSITION).toString()) - 1);
-
- } else if ((nameObj.get(Constants.COLUMN_NAME)) != null) {
-
- nameDef.columnName = (String) (nameObj.get(Constants.COLUMN_NAME));
-
+ JsonNode posNode = nameNode.get(Constants.COLUMN_POSITION);
+ if (posNode != null) {
+ nameDef.columnPos = (Integer.parseInt(posNode.asText()) - 1);
} else {
- log.error("Column_name or pos info is missing. Specify proper bin name mapping in config file for: " + jobj.toString());
+ JsonNode colNameNode = nameNode.get(Constants.COLUMN_NAME);
+ if (colNameNode != null) {
+ nameDef.columnName = colNameNode.asText();
+ } else {
+ log.error("Column_name or pos info is missing. Specify proper bin name mapping in config file for: " + RelaxedJsonMapper.jsonNodeToString(jobj));
+ }
}
}
@@ -314,33 +301,40 @@ private static BinDefinition getBinDefs(JSONObject jobj) {
ColumnDefinition valueDef = new ColumnDefinition(-1, null, null, null, null, null, null);
String staticBinValue = null;
- if ((obj = jobj.get(Constants.VALUE)) == null) {
- log.error(Constants.VALUE + " key is missing in bin object:" + jobj.toString());
+ JsonNode valueNode = jobj.get(Constants.VALUE);
+ if (valueNode == null) {
+ log.error(Constants.VALUE + " key is missing in bin object:" + RelaxedJsonMapper.jsonNodeToString(jobj));
return null;
- } else if (!(obj instanceof JSONObject)) {
- staticBinValue = (String) obj;
+ } else if (valueNode.isTextual()) {
+ staticBinValue = valueNode.asText();
} else {
- JSONObject valueObj = (JSONObject) obj;
-
- if ((valueObj.get(Constants.COLUMN_POSITION)) != null) {
-
- valueDef.columnPos = (Integer.parseInt(valueObj.get(Constants.COLUMN_POSITION).toString()) - 1);
-
- } else if ((valueObj.get(Constants.COLUMN_NAME)) != null) {
-
- valueDef.columnName = (String) (valueObj.get(Constants.COLUMN_NAME));
-
+ JsonNode posNode = valueNode.get(Constants.COLUMN_POSITION);
+ if (posNode != null) {
+ valueDef.columnPos = (Integer.parseInt(posNode.asText()) - 1);
} else {
- log.error("Column_name or pos info is missing. Specify proper bin value mapping in config file for: " + jobj.toString());
+ JsonNode colNameNode = valueNode.get(Constants.COLUMN_NAME);
+ if (colNameNode != null) {
+ valueDef.columnName = colNameNode.asText();
+
+ } else {
+ log.error("Column_name or pos info is missing. Specify proper bin value mapping in config file for: " + RelaxedJsonMapper.jsonNodeToString(jobj));
+ }
}
- valueDef.setSrcType((String) (valueObj.get(Constants.TYPE)));
+ JsonNode typeNode = valueNode.get(Constants.TYPE);
+ valueDef.setSrcType(typeNode != null ? typeNode.asText() : null);
if (valueDef.srcType == null) {
- log.error(Constants.TYPE + " key is missing in bin object: " + jobj.toString());
+ log.error(Constants.TYPE + " key is missing in bin object: " + RelaxedJsonMapper.jsonNodeToString(jobj));
}
- valueDef.setDstType((String) (valueObj.get(Constants.DST_TYPE)));
- valueDef.encoding = (String) (valueObj.get(Constants.ENCODING));
- valueDef.removePrefix = ((String) (valueObj.get(Constants.REMOVE_PREFIX)));
+
+ JsonNode dstTypeNode = valueNode.get(Constants.DST_TYPE);
+ valueDef.setDstType(dstTypeNode != null ? dstTypeNode.asText() : null);
+
+ JsonNode encodingNode = valueNode.get(Constants.ENCODING);
+ valueDef.encoding = encodingNode != null ? encodingNode.asText() : null;
+
+ JsonNode prefixNode = valueNode.get(Constants.REMOVE_PREFIX);
+ valueDef.removePrefix = prefixNode != null ? prefixNode.asText() : null;
}
return new BinDefinition(staticBinName, staticBinValue, nameDef, valueDef);
diff --git a/src/main/java/com/aerospike/load/RelaxedJsonMapper.java b/src/main/java/com/aerospike/load/RelaxedJsonMapper.java
new file mode 100644
index 0000000..97769cf
--- /dev/null
+++ b/src/main/java/com/aerospike/load/RelaxedJsonMapper.java
@@ -0,0 +1,221 @@
+package com.aerospike.load;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Utility class for parsing JSON with relaxed syntax using Jackson.
+ * Supports unquoted field names, single quotes, and automatic type coercion.
+ */
+public class RelaxedJsonMapper {
+
+ private static final ObjectMapper RELAXED_MAPPER = new ObjectMapper();
+ private static final ObjectMapper STANDARD_MAPPER = new ObjectMapper();
+ private static final Logger log = LogManager.getLogger(RelaxedJsonMapper.class);
+
+ static {
+ // Configure relaxed mapper to allow JSON supersets
+ RELAXED_MAPPER.enable(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES);
+ RELAXED_MAPPER.enable(JsonParser.Feature.ALLOW_SINGLE_QUOTES);
+ RELAXED_MAPPER.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
+ // Use regular integers instead of BigInteger for better compatibility
+ RELAXED_MAPPER.disable(DeserializationFeature.USE_BIG_INTEGER_FOR_INTS);
+ }
+
+ /**
+ * Parse JSON from a FileReader.
+ * @param reader FileReader containing JSON data
+ * @return JsonNode representing the parsed JSON
+ * @throws IOException if parsing fails
+ */
+ public static JsonNode parseJson(FileReader reader) throws IOException {
+ return RELAXED_MAPPER.readTree(reader);
+ }
+
+ /**
+ * Coerce a string key to its appropriate type (Integer, Long, Double, Boolean, or String).
+ * @param key The string key to coerce
+ * @return The coerced key object
+ */
+ private static Object coerceKey(String key) {
+ if (key == null) {
+ return key;
+ }
+
+ // Try to parse as boolean
+ if ("true".equalsIgnoreCase(key)) {
+ return true;
+ }
+ if ("false".equalsIgnoreCase(key)) {
+ return false;
+ }
+
+ // Try to parse as integer
+ try {
+ if (!key.contains(".") && !key.contains("e") && !key.contains("E")) {
+ long longValue = Long.parseLong(key);
+ if (longValue >= Integer.MIN_VALUE && longValue <= Integer.MAX_VALUE) {
+ return (int) longValue;
+ } else {
+ return longValue;
+ }
+ }
+ } catch (NumberFormatException e) {
+ // Not an integer
+ }
+
+ // Try to parse as double
+ try {
+ return Double.parseDouble(key);
+ } catch (NumberFormatException e) {
+ // Not a double
+ }
+
+ // Return as string
+ return key;
+ }
+
+ /**
+ * Parse JSON string and return appropriate type based on content.
+ * For arrays: returns List