Fix auto column mapping bug when target has more columns than origin (#375)

pravinbhat · msmygit · web-flow · commit 7894842021a8 · 2025-06-27T10:53:40.000-04:00
* Convert nulls on origin to unset at target by a new property

* Undo the option to allow null values as they will be eliminated during compaction. It provides no value other than slowing migration &amp; target cluster.
Also, doing a strict check for String (which includes all C* text types) types with empty values

* Added release notes and updated readme

* Added missing license

* Defensive code to avoid UNSET exceptions

* Handle bind counters correctly when target has more columns than origin

* Added tests for auto column mapping test &amp; DiffData bug-fix

* Added release notes

---------

Co-authored-by: Madhavan Sridharan &lt;madhavan.sridharan@datastax.com&gt;
diff --git a/RELEASE.md b/RELEASE.md
@@ -1,5 +1,8 @@
 # Release Notes
 
+## [5.4.1] - 2025-06-26
+- Bug fix: Fixed auto column mapping bug when `target` table has more columns than `origin`.
+
 ## [5.4.0] - 2025-06-16
 - Use `UNSET` value for null fields (including empty texts) to avoid creating (or carrying forward) tombstones during row creation.
 
diff --git a/SIT/features/08_map_columns_origin_target/breakData.cql b/SIT/features/08_map_columns_origin_target/breakData.cql
@@ -0,0 +1,23 @@
+/*
+ Licensed 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.
+*/
+
+DELETE FROM target.map_columns WHERE key_a=2;
+UPDATE target.map_columns SET val_a='valueD' WHERE key_a=3;
+
+INSERT INTO origin.map_columns(key_a, key_b, val_a, val_b) VALUES (1, 'key1','valueA', 21);
+INSERT INTO origin.map_columns(key_a, key_b, val_a, val_b) VALUES (2, 'key2','valueB', 22);
+INSERT INTO origin.map_columns(key_a, key_b, val_a, val_b) VALUES (3, 'key3','valueC', 23);
+
+SELECT * FROM target.map_columns;
+
diff --git a/SIT/features/08_map_columns_origin_target/cdm.fixData.assert b/SIT/features/08_map_columns_origin_target/cdm.fixData.assert
@@ -0,0 +1,8 @@
+Read Record Count: 3
+Mismatch Record Count: 1
+Corrected Mismatch Record Count: 1
+Missing Record Count: 1
+Corrected Missing Record Count: 1
+Valid Record Count: 1
+Skipped Record Count: 0
+Error Record Count: 0
diff --git a/SIT/features/08_map_columns_origin_target/cdm.migrateData.assert b/SIT/features/08_map_columns_origin_target/cdm.migrateData.assert
@@ -0,0 +1,4 @@
+Read Record Count: 3
+Write Record Count: 3
+Skipped Record Count: 0
+Error Record Count: 0
diff --git a/SIT/features/08_map_columns_origin_target/cdm.txt b/SIT/features/08_map_columns_origin_target/cdm.txt
@@ -0,0 +1,3 @@
+migrateData com.datastax.cdm.job.Migrate migrate.properties
+validateData com.datastax.cdm.job.DiffData migrate.properties
+fixData com.datastax.cdm.job.DiffData fix.properties
diff --git a/SIT/features/08_map_columns_origin_target/cdm.validateData.assert b/SIT/features/08_map_columns_origin_target/cdm.validateData.assert
@@ -0,0 +1,8 @@
+Read Record Count: 3
+Mismatch Record Count: 0
+Corrected Mismatch Record Count: 0
+Missing Record Count: 0
+Corrected Missing Record Count: 0
+Valid Record Count: 3
+Skipped Record Count: 0
+Error Record Count: 0
diff --git a/SIT/features/08_map_columns_origin_target/execute.sh b/SIT/features/08_map_columns_origin_target/execute.sh
@@ -0,0 +1,29 @@
+#
+# Licensed 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.
+#
+
+#!/bin/bash -e
+
+workingDir="$1"
+cd "$workingDir"
+
+/local/cdm.sh -f cdm.txt -s migrateData -d "$workingDir" > cdm.migrateData.out 2>cdm.migrateData.err
+/local/cdm-assert.sh -f cdm.migrateData.out -a cdm.migrateData.assert -d "$workingDir"
+
+/local/cdm.sh -f cdm.txt -s validateData -d "$workingDir" > cdm.validateData.out 2>cdm.validateData.err
+/local/cdm-assert.sh -f cdm.validateData.out -a cdm.validateData.assert -d "$workingDir"
+
+cqlsh -u $CASS_USERNAME -p $CASS_PASSWORD $CASS_CLUSTER -f $workingDir/breakData.cql > $workingDir/breakData.out 2> $workingDir/breakData.err
+
+/local/cdm.sh -f cdm.txt -s fixData -d "$workingDir" > cdm.fixData.out 2>cdm.fixData.err
+/local/cdm-assert.sh -f cdm.fixData.out -a cdm.fixData.assert -d "$workingDir"
diff --git a/SIT/features/08_map_columns_origin_target/expected.cql b/SIT/features/08_map_columns_origin_target/expected.cql
@@ -0,0 +1,15 @@
+/*
+ Licensed 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.
+*/
+
+SELECT * FROM target.map_columns;
diff --git a/SIT/features/08_map_columns_origin_target/expected.out b/SIT/features/08_map_columns_origin_target/expected.out
@@ -0,0 +1,8 @@
+
+ key_a | val_a  | val_c
+-------+--------+-------
+     1 | valueA |  null
+     2 | valueB |  null
+     3 | valueC |  null
+
+(3 rows)
diff --git a/SIT/features/08_map_columns_origin_target/fix.properties b/SIT/features/08_map_columns_origin_target/fix.properties
@@ -0,0 +1,23 @@
+#
+# Licensed 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.
+#
+
+spark.cdm.connect.origin.host                     cdm-sit-cass
+spark.cdm.connect.target.host                     cdm-sit-cass
+
+spark.cdm.schema.origin.keyspaceTable             origin.map_columns
+spark.cdm.schema.target.keyspaceTable             target.map_columns
+spark.cdm.perfops.numParts                        1
+
+spark.cdm.autocorrect.missing                     true
+spark.cdm.autocorrect.mismatch                    true
diff --git a/SIT/features/08_map_columns_origin_target/migrate.properties b/SIT/features/08_map_columns_origin_target/migrate.properties
@@ -0,0 +1,20 @@
+#
+# Licensed 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.
+#
+
+spark.cdm.connect.origin.host                     cdm-sit-cass
+spark.cdm.connect.target.host                     cdm-sit-cass
+
+spark.cdm.schema.origin.keyspaceTable             origin.map_columns
+spark.cdm.schema.target.keyspaceTable             target.map_columns
+spark.cdm.perfops.numParts                        1
diff --git a/SIT/features/08_map_columns_origin_target/setup.cql b/SIT/features/08_map_columns_origin_target/setup.cql
@@ -0,0 +1,22 @@
+/*
+ Licensed 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.
+*/
+
+DROP TABLE IF EXISTS origin.map_columns;
+CREATE TABLE origin.map_columns(key_a int, key_b text, val_a text, val_b int, PRIMARY KEY (key_a, key_b));
+INSERT INTO origin.map_columns(key_a, key_b, val_a, val_b) VALUES (1, 'key1','valueA', 21);
+INSERT INTO origin.map_columns(key_a, key_b, val_a, val_b) VALUES (2, 'key2','valueB', 22);
+INSERT INTO origin.map_columns(key_a, key_b, val_a, val_b) VALUES (3, 'key3','valueC', 23);
+
+DROP TABLE IF EXISTS target.map_columns;
+CREATE TABLE target.map_columns(key_a int, val_a text, val_c int, PRIMARY KEY (key_a));
diff --git a/src/main/java/com/datastax/cdm/cql/statement/TargetInsertStatement.java b/src/main/java/com/datastax/cdm/cql/statement/TargetInsertStatement.java
@@ -72,16 +72,18 @@ protected BoundStatement bind(Row originRow, Row targetRow, Integer ttl, Long wr
                 } else {
                     int originIndex = cqlTable.getCorrespondingIndex(targetIndex);
                     if (originIndex < 0) // we don't have data to bind for this column; continue to the next targetIndex
+                    {
+                        currentBindIndex++;
                         continue;
+                    }
                     bindValue = cqlTable.getOtherCqlTable().getAndConvertData(originIndex, originRow);
                 }
 
-                if (null == bindValue || bindValue instanceof String && ((String) bindValue).isEmpty()) {
-                    boundStatement = boundStatement.unset(currentBindIndex++);
-                } else {
-                    boundStatement = boundStatement.set(currentBindIndex++, bindValue,
+                if (!(null == bindValue || (bindValue instanceof String && ((String) bindValue).isEmpty()))) {
+                    boundStatement = boundStatement.set(currentBindIndex, bindValue,
                             cqlTable.getBindClass(targetIndex));
                 }
+                currentBindIndex++;
             } catch (Exception e) {
                 logger.error(
                         "Error trying to bind value: {} of class: {} to column: {} of targetDataType: {}/{} at column index: {} and bind index: {} of statement: {}",
diff --git a/src/main/java/com/datastax/cdm/cql/statement/TargetUpdateStatement.java b/src/main/java/com/datastax/cdm/cql/statement/TargetUpdateStatement.java
@@ -81,19 +81,20 @@ protected BoundStatement bind(Row originRow, Row targetRow, Integer ttl, Long wr
                     originIndex = extractJsonFeature.getOriginColumnIndex();
                     bindValueTarget = extractJsonFeature.extract(originRow.getString(originIndex));
                 } else {
-                    if (originIndex < 0)
-                        // we don't have data to bind for this column; continue to the next targetIndex
+                    if (originIndex < 0) // we don't have data to bind for this column; continue to the next targetIndex
+                    {
+                        currentBindIndex++;
                         continue;
+                    }
                     bindValueTarget = cqlTable.getOtherCqlTable().getAndConvertData(originIndex, originRow);
                 }
 
-                if (null == bindValueTarget
-                        || bindValueTarget instanceof String && ((String) bindValueTarget).isEmpty()) {
-                    boundStatement = boundStatement.unset(currentBindIndex++);
-                } else {
-                    boundStatement = boundStatement.set(currentBindIndex++, bindValueTarget,
+                if (!(null == bindValueTarget
+                        || (bindValueTarget instanceof String && ((String) bindValueTarget).isEmpty()))) {
+                    boundStatement = boundStatement.set(currentBindIndex, bindValueTarget,
                             cqlTable.getBindClass(targetIndex));
                 }
+                currentBindIndex++;
             } catch (Exception e) {
                 logger.error("Error trying to bind value:" + bindValueTarget + " to column:"
                         + targetColumnNames.get(targetIndex) + " of targetDataType:"
diff --git a/src/main/java/com/datastax/cdm/job/DiffJobSession.java b/src/main/java/com/datastax/cdm/job/DiffJobSession.java
@@ -290,12 +290,6 @@ private String isDifferent(Record record) {
                                         pk, targetIndex, targetColumnNames.get(targetIndex), origin);
                         } else if (targetIndex == extractJsonFeature.getTargetColumnIndex()) {
                             originIndex = extractJsonFeature.getOriginColumnIndex();
-                        } else {
-                            throw new RuntimeException("Target column \"" + targetColumnNames.get(targetIndex)
-                                    + "\" at index " + targetIndex
-                                    + " cannot be found on Origin, and is neither a constant column (indexes:"
-                                    + constantColumnIndexes + ") nor an explode map column (keyIndex:"
-                                    + explodeMapKeyIndex + ", valueIndex:" + explodeMapValueIndex + ")");
                         }
                     }
 
diff --git a/src/test/java/com/datastax/cdm/cql/statement/TargetInsertStatementTest.java b/src/test/java/com/datastax/cdm/cql/statement/TargetInsertStatementTest.java
@@ -234,7 +234,6 @@ public void bind_nullToUnset_true() {
 
         // Assert
         assertNotNull(result);
-        verify(boundStatement, atLeastOnce()).unset(anyInt());
     }
 
 }
diff --git a/src/test/java/com/datastax/cdm/cql/statement/TargetUpdateStatementTest.java b/src/test/java/com/datastax/cdm/cql/statement/TargetUpdateStatementTest.java
@@ -212,7 +212,6 @@ public void bind_nullToUnset_true() {
 
         // Assert
         assertNotNull(result);
-        verify(boundStatement, atLeastOnce()).unset(anyInt());
     }
 
 }

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+migrateData com.datastax.cdm.job.Migrate migrate.properties`
	`2`	`+validateData com.datastax.cdm.job.DiffData migrate.properties`
	`3`	`+fixData com.datastax.cdm.job.DiffData fix.properties`