[BACKLOG-26374] steps creating field name conflicts (#5939)

repeated field names could arise after inserting or deleting value metas
select values not checking for conflicts when renaming

core/src/main/java/org/pentaho/di/core/row/RowMeta.java

@@ -253,7 +253,8 @@ public void addValueMeta( int index, ValueMetaInterface meta ) {
           newMeta = renameValueMetaIfInRow( meta, null );
         }
         valueMetaList.add( index, newMeta );
-        cache.invalidate();
+        cache.storeMapping( newMeta.getName(), index );
+        cache.updateFrom( index + 1, valueMetaList );
         needRealClone = null;
       } finally {
         lock.writeLock().unlock();
@@ -809,8 +810,11 @@ public void removeValueMeta( String valueName ) throws KettleValueException {
   public void removeValueMeta( int index ) {
     lock.writeLock().lock();
     try {
-      valueMetaList.remove( index );
-      cache.invalidate();
+      ValueMetaInterface old = valueMetaList.remove( index );
+      if ( old != null ) {
+        cache.removeMapping( old.getName() );
+        cache.updateFrom( index, valueMetaList );
+      }
       needRealClone = null;
     } finally {
       lock.writeLock().unlock();
@@ -1291,12 +1295,22 @@ void storeMapping( String name, int index ) {
     }
 
     void replaceMapping( String old, String current, int index ) {
-      if ( !Utils.isEmpty( old ) ) {
-        mapping.remove( old.toLowerCase() );
-      }
+      removeMapping( old );
       storeMapping( current, index );
     }
 
+    void removeMapping( String toRemove ) {
+      if ( !Utils.isEmpty( toRemove ) ) {
+        mapping.remove( toRemove.toLowerCase() );
+      }
+    }
+
+    void updateFrom( int idx, List<ValueMetaInterface> metas ) {
+      for ( int i = idx; i < metas.size(); i++  ) {
+        storeMapping( metas.get( i ).getName(), i );
+      }
+    }
+
     Integer findAndCompare( String name, List<? extends ValueMetaInterface> metas ) {
       if ( Utils.isEmpty( name ) ) {
         return null;

core/src/test/java/org/pentaho/di/core/row/RowMetaTest.java

@@ -214,6 +214,23 @@ public void testSetValueMetaDup() throws KettlePluginException {
     assertEquals( "Original is still the same (name)", 1, rowMeta.indexOfValue( "dup" ) );
     assertEquals( "Renaming happened", 2, rowMeta.indexOfValue( "dup_1" ) );
   }
+
+  @Test
+  public void testInsertValueMetaDup() throws Exception {
+    rowMeta.addValueMeta( 1, new ValueMetaInteger( integer.getName() ) );
+    assertEquals( "inserted", 4, rowMeta.size() );
+    assertEquals( "rename new", "integer_1", rowMeta.getValueMeta( 1 ).getName() );
+    rowMeta.addValueMeta( new ValueMetaInteger( integer.getName() ) );
+    assertEquals( "rename after", "integer_2", rowMeta.getValueMeta( 4 ).getName() );
+  }
+
+  @Test
+  public void testRemoveValueMetaDup() throws Exception {
+    rowMeta.removeValueMeta( date.getName() );
+    assertEquals( "removed", 2, rowMeta.size() );
+    rowMeta.addValueMeta( new ValueMetaInteger( integer.getName() ) );
+    assertEquals( "rename after", "integer_1", rowMeta.getValueMeta( 2 ).getName() );
+  }
 
   @Test
   public void testSetValueMetaNullName() throws KettlePluginException {

engine/src/main/java/org/pentaho/di/trans/steps/selectvalues/SelectValuesMeta.java

@@ -411,6 +411,8 @@ public void getMetadataFields( RowMetaInterface inputRowMeta, String name, Varia
           if ( !v.getName().equals( metaChange.getRename() ) && !Utils.isEmpty( metaChange.getRename() ) ) {
             v.setName( metaChange.getRename() );
             v.setOrigin( name );
+            // need to reinsert to check name conflicts
+            inputRowMeta.setValueMeta( idx, v );
           }
           // Change the type?
           if ( metaChange.getType() != ValueMetaInterface.TYPE_NONE && v.getType() != metaChange.getType() ) {

engine/src/test/java/org/pentaho/di/trans/steps/selectvalues/SelectValuesMetaTest.java

@@ -2,7 +2,7 @@
  *
  * Pentaho Data Integration
  *
- * Copyright (C) 2002-2017 by Hitachi Vantara : http://www.pentaho.com
+ * Copyright (C) 2002-2018 by Hitachi Vantara : http://www.pentaho.com
  *
  *******************************************************************************
  *
@@ -36,6 +36,9 @@
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.pentaho.di.core.exception.KettleException;
+import org.pentaho.di.core.row.RowMeta;
+import org.pentaho.di.core.row.RowMetaInterface;
+import org.pentaho.di.core.row.value.ValueMetaString;
 import org.pentaho.di.junit.rules.RestorePDIEngineEnvironment;
 import org.pentaho.di.trans.steps.loadsave.LoadSaveTester;
 import org.pentaho.di.trans.steps.loadsave.validator.ArrayLoadSaveValidator;
@@ -71,8 +74,8 @@ public void loadSaveTest() throws KettleException {
     fieldLoadSaveValidatorTypeMap.put( SelectField[].class.getCanonicalName(), new ArrayLoadSaveValidator<SelectField>(
         new SelectFieldLoadSaveValidator( selectField ), 2 ) );
 
-    LoadSaveTester tester =
-        new LoadSaveTester( SelectValuesMeta.class, attributes, new HashMap<String, String>(),
+    LoadSaveTester<SelectValuesMeta> tester =
+        new LoadSaveTester<>( SelectValuesMeta.class, attributes, new HashMap<String, String>(),
             new HashMap<String, String>(), new HashMap<String, FieldLoadSaveValidator<?>>(),
             fieldLoadSaveValidatorTypeMap );
 
@@ -210,6 +213,21 @@ public void getSelectPrecision() {
     assertArrayEquals( new int[0], selectValuesMeta.getSelectPrecision() );
   }
 
+  @Test
+  public void testMetaDataFieldsRenameConflict() throws Exception {
+    RowMetaInterface rowMeta = new RowMeta();
+    rowMeta.addValueMeta( new ValueMetaString( "A" ) );
+    rowMeta.addValueMeta( new ValueMetaString( "B" ) );
+
+    SelectMetadataChange change = new SelectMetadataChange( selectValuesMeta );
+    change.setName( "A" );
+    change.setRename( "B" );
+    selectValuesMeta.setMeta( new SelectMetadataChange[] { change } );
+
+    selectValuesMeta.getMetadataFields( rowMeta, "select values", null );
+    assertEquals( "rename conflict", "B_1", rowMeta.getValueMeta( 0 ).getName() );
+  }
+
   public static class SelectFieldLoadSaveValidator implements FieldLoadSaveValidator<SelectField> {
 
     private final SelectField defaultValue;

engine/src/test/java/org/pentaho/di/trans/steps/selectvalues/SelectValuesTest.java

@@ -30,7 +30,6 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
-import org.junit.runner.RunWith;
 import org.pentaho.di.core.Const;
 import org.pentaho.di.core.KettleEnvironment;
 import org.pentaho.di.core.RowSet;