Fixed empty value in map issue

Author: pravinbhat

.classpath

@@ -9,16 +9,17 @@
 	<classpathentry excluding="**" kind="src" output="target/classes" path="src/resources">
 		<attributes>
 			<attribute name="maven.pomderived" value="true"/>
+			<attribute name="optional" value="true"/>
 		</attributes>
 	</classpathentry>
 	<classpathentry kind="src" output="target/test-classes" path="src/test/java">
 		<attributes>
+			<attribute name="test" value="true"/>
 			<attribute name="optional" value="true"/>
 			<attribute name="maven.pomderived" value="true"/>
-			<attribute name="test" value="true"/>
 		</attributes>
 	</classpathentry>
-	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.8">
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-11">
 		<attributes>
 			<attribute name="maven.pomderived" value="true"/>
 		</attributes>
@@ -28,5 +29,23 @@
 			<attribute name="maven.pomderived" value="true"/>
 		</attributes>
 	</classpathentry>
+	<classpathentry excluding="**" kind="src" output="target/test-classes" path="src/test/resources">
+		<attributes>
+			<attribute name="test" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+			<attribute name="optional" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="src" path="target/generated-sources/annotations">
+		<attributes>
+			<attribute name="optional" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="src" output="target/test-classes" path="target/generated-test-sources/test-annotations">
+		<attributes>
+			<attribute name="optional" value="true"/>
+			<attribute name="test" value="true"/>
+		</attributes>
+	</classpathentry>
 	<classpathentry kind="output" path="target/classes"/>
 </classpath>

RELEASE.md

@@ -1,4 +1,8 @@
 # Release Notes
+## [4.1.16] - 2024-05-31
+- Added property to manage null values in Map fields
+- Updated README
+  
 ## [4.1.15] - 2024-03-05
 - Internal CI/CD release fix
 

src/main/java/com/datastax/cdm/feature/WritetimeTTL.java

@@ -147,11 +147,12 @@ public boolean initializeAndValidate(CqlTable originTable, CqlTable targetTable)
             isValid = false;
         }
 
-        if (this.writetimeIncrement == 0L && (null!=writetimeNames && !writetimeNames.isEmpty()) && originTable.hasUnfrozenList()) {
-            logger.warn("Writetime is configured, but the origin table at least one unfrozen List, and there is a zero-value increment configured at "+
-                    KnownProperties.TRANSFORM_CUSTOM_WRITETIME_INCREMENT+"; this may result in duplicate list entries when "+
-                    KnownProperties.AUTOCORRECT_MISMATCH+" is enabled.");
-        }
+		if (this.writetimeIncrement == 0L && (null != writetimeNames && !writetimeNames.isEmpty())
+				&& originTable.hasUnfrozenList()) {
+			logger.warn("Origin table has at least one unfrozen List, and "
+					+ KnownProperties.TRANSFORM_CUSTOM_WRITETIME_INCREMENT
+					+ " is set to zero; this may result in duplicate list entries on reruns or validation with autocorrect.");
+		}
 
         if (!isValid) isEnabled = false;
         logger.info("Feature {} is {}", this.getClass().getSimpleName(), isEnabled?"enabled":"disabled");

src/main/java/com/datastax/cdm/properties/KnownProperties.java

@@ -15,7 +15,13 @@
  */
 package com.datastax.cdm.properties;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 public class KnownProperties {
 
@@ -167,24 +173,27 @@ public enum PropertyType {
     public static final String TRANSFORM_CODECS                                = "spark.cdm.transform.codecs";
     public static final String TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT        = "spark.cdm.transform.codecs.timestamp.string.format";
     public static final String TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT_ZONE   = "spark.cdm.transform.codecs.timestamp.string.zone";
+    public static final String TRANSFORM_MAP_REMOVE_KEY_WITH_NO_VALUE          = "spark.cdm.transform.map.remove.null.vaue";
 
 
 
 //    TODO: 3.3.0 refactored how guardrails are handled, this needs to be merged forward
 //    public static final String GUARDRAIL_FIELD_LIMIT_MB         = "spark.guardrail.colSizeInKB"; //10
 
-    static {
-        types.put(TRANSFORM_REPLACE_MISSING_TS, PropertyType.NUMBER);
-        types.put(TRANSFORM_CUSTOM_WRITETIME, PropertyType.NUMBER);
-     defaults.put(TRANSFORM_CUSTOM_WRITETIME, "0");
-        types.put(TRANSFORM_CUSTOM_WRITETIME_INCREMENT, PropertyType.NUMBER);
-     defaults.put(TRANSFORM_CUSTOM_WRITETIME_INCREMENT, "0");
-        types.put(TRANSFORM_CODECS, PropertyType.STRING_LIST);
-        types.put(TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT, PropertyType.STRING);
-     defaults.put(TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT, "yyyyMMddHHmmss");
-        types.put(TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT_ZONE, PropertyType.STRING);
-     defaults.put(TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT_ZONE, "UTC");
-    }
+	static {
+		types.put(TRANSFORM_REPLACE_MISSING_TS, PropertyType.NUMBER);
+		types.put(TRANSFORM_CUSTOM_WRITETIME, PropertyType.NUMBER);
+		defaults.put(TRANSFORM_CUSTOM_WRITETIME, "0");
+		types.put(TRANSFORM_CUSTOM_WRITETIME_INCREMENT, PropertyType.NUMBER);
+		defaults.put(TRANSFORM_CUSTOM_WRITETIME_INCREMENT, "0");
+		types.put(TRANSFORM_CODECS, PropertyType.STRING_LIST);
+		types.put(TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT, PropertyType.STRING);
+		defaults.put(TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT, "yyyyMMddHHmmss");
+		types.put(TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT_ZONE, PropertyType.STRING);
+		defaults.put(TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT_ZONE, "UTC");
+		types.put(TRANSFORM_MAP_REMOVE_KEY_WITH_NO_VALUE, PropertyType.BOOLEAN);
+		defaults.put(TRANSFORM_MAP_REMOVE_KEY_WITH_NO_VALUE, "false");
+	}
 
     //==========================================================================
     // Cassandra-side Filters

src/main/java/com/datastax/cdm/schema/CqlTable.java

@@ -15,9 +15,29 @@
  */
 package com.datastax.cdm.schema;
 
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.cdm.data.CqlConversion;
+import com.datastax.cdm.data.CqlData;
+import com.datastax.cdm.data.DataUtility;
 import com.datastax.cdm.feature.Feature;
 import com.datastax.cdm.feature.Featureset;
 import com.datastax.cdm.feature.WritetimeTTL;
+import com.datastax.cdm.properties.KnownProperties;
+import com.datastax.cdm.properties.PropertyHelper;
 import com.datastax.oss.driver.api.core.ConsistencyLevel;
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.CqlSession;
@@ -26,22 +46,12 @@
 import com.datastax.oss.driver.api.core.metadata.schema.ColumnMetadata;
 import com.datastax.oss.driver.api.core.metadata.schema.KeyspaceMetadata;
 import com.datastax.oss.driver.api.core.metadata.schema.TableMetadata;
-import com.datastax.oss.driver.api.core.type.*;
+import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.DataTypes;
+import com.datastax.oss.driver.api.core.type.ListType;
+import com.datastax.oss.driver.api.core.type.TupleType;
 import com.datastax.oss.driver.api.core.type.codec.CodecNotFoundException;
 import com.datastax.oss.driver.api.core.type.codec.registry.MutableCodecRegistry;
-import com.datastax.cdm.data.CqlData;
-import com.datastax.cdm.data.CqlConversion;
-import com.datastax.cdm.data.DataUtility;
-import com.datastax.cdm.properties.KnownProperties;
-import com.datastax.cdm.properties.PropertyHelper;
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.time.Instant;
-import java.util.*;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
 
 public class CqlTable extends BaseTable {
     public Logger logger = LoggerFactory.getLogger(this.getClass().getName());
@@ -163,6 +173,9 @@ public Integer getBatchSize() {
             return prop;
     }
 
+	private boolean removeMapWithNoValues = propertyHelper
+			.getBoolean(KnownProperties.TRANSFORM_MAP_REMOVE_KEY_WITH_NO_VALUE);
+
     // Adds to the current column list based on the name and type of columns already existing in the table
     // This is useful where a feature is adding a column by name of an existing column.
     // If the column is already present, the bind class is added to the return list.
@@ -277,6 +290,11 @@ public Object getAndConvertData(int index, Row row) {
         if (null==thisObject) {
             return convertNull(index);
         }
+
+		if (removeMapWithNoValues && thisObject instanceof Map) {
+			removeNullValuesFrmMap(thisObject);
+		}
+
         CqlConversion cqlConversion = this.cqlConversions.get(index);
         if (null==cqlConversion) {
             if (logTrace) logger.trace("{} Index:{} not converting:{}",isOrigin?"origin":"target",index,thisObject);
@@ -288,6 +306,12 @@ public Object getAndConvertData(int index, Row row) {
         }
     }
 
+	private Object removeNullValuesFrmMap(Object thisObject) {
+		Set<Map.Entry> ms = (((Map) thisObject).entrySet());
+		return ms.stream().filter(e -> (e.getValue() != null))
+				.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+	}
+
     public Object convertNull(int thisIndex) {
         // We do not need to convert nulls for non-PK columns
         int otherIndex = this.getCorrespondingIndex(thisIndex);

src/resources/cdm-detailed.properties

@@ -170,13 +170,16 @@ spark.cdm.autocorrect.mismatch                    false
 #  spark.cdm.perfops
 #    .numParts            : Defaults is 10000. In standard operation, the full token range (-2^63..2^63-1)  
 #                           is divided into a number of parts which will be parallel-processed. You should 
-#                           aim for each part to comprise a total of ≈1-10GB of data to migrate. During 
-#                           initial testing, you may want this to be a small number (even 1).
+#                           aim for each part to be ~10MB of data to migrate. The default is assuming the 
+#                           table size is close to 100GB (10MB * 10K), if the table size is significantly 
+#                           less or more than that, adjust this value accordingly.
 #    .batchSize           : Defaults is 5. When writing to Target, this comprises the number of records that 
 #                           will be put into an UNLOGGED batch. CDM will tend to work on the same partition 
 #                           at a time so if your partition sizes are larger, this number may be increased. 
 #                           If .batchSize would mean that more than 1 partition is often contained in a batch,  
 #                           the figure should be reduced. Ideally < 1% of batches have more than 1 partition.
+#                           For tables where primary-key=partition-key OR average row-size larger than 20 KB,
+#                           always set this value to 1.
 #    .ratelimit
 #      .origin            : Defaults to 20000. Concurrent number of operations across all parallel threads
 #                           from Origin. This may be adjusted up (or down), depending on the amount of data
@@ -253,13 +256,19 @@ spark.cdm.perfops.ratelimit.target                40000
 #  .codecs.timestamp               Configuration for CQL_TIMESTAMP_TO_STRING_FORMAT codec.
 #     .string.format               Default is yyyyMMddHHmmss ; DateTimeFormatter.ofPattern(formatString)
 #     .string.zone                 Default is UTC ; Must be in ZoneRulesProvider.getAvailableZoneIds()
+#
+#  .map.remove.null.value          Default is false. Setting this to true will remove any entries in Map 
+#                                  field with empty or null values. Such values can create NPE exception
+#                                  if the value type does not support empty or null values (like Timestamp)
+#                                  and this property is false. Set it to true to handle such exceptions.
 #-----------------------------------------------------------------------------------------------------------
 #spark.cdm.transform.missing.key.ts.replace.value      1685577600000
 #spark.cdm.transform.custom.writetime                  0
 #spark.cdm.transform.custom.writetime.incrementBy      0
 #spark.cdm.transform.codecs
 #spark.cdm.transform.codecs.timestamp.string.format    yyyyMMddHHmmss
 #spark.cdm.transform.codecs.timestamp.string.zone      UTC
+#spark.cdm.transform.map.remove.null.vaue              false
 
 #===========================================================================================================
 # Cassandra Filters are applied on the coordinator node. Note that, depending on the filter, the coordinator