Fix ALTER TABLE .. ADD to work if table contains nested primary keys

This fixes #9386

`AnalyzedTableElements.toMapping()` requires the individual table
elements to be connected with their child elements to correctly generate
the mapping for object columns.

Internally when we process a `ALTER TABLE .. ADD` we implicitly re-add
all existing primary keys to the `AnalyzedTableElements` structure, but
if the primary keys were nested columns we didn't build the object
  graph.

Furthermore, we first used the `ident(..)` setter followed by the
`name(..)` setter, the `name(..)` setter implicitly created the `ident`
which meant that all nested primary keys were changed to their top-level
object name.

This caused a `column "xy" specified more than once` error.

(cherry picked from commit 717dcd3)

docs/appendices/release-notes/unreleased.rst

@@ -46,6 +46,9 @@ None
 Fixes
 =====
 
+- Fixed an issue that caused an error when using ``ALTER TABLE .. ADD`` on a
+  table which contains nested primary key columns.
+
 - Fixed an issue where values of type ``array(varchar)`` were decoded
   incorrectly if they contained a ``,`` character. This occurred when
   the PostgreSQL wire protocol was used in ``text`` mode.
@@ -54,4 +57,4 @@ Fixes
   introduced a performance regression on the snapshot process.
 
 - Fixed a ``ClassCastException`` that could occur when using ``unnest`` on
-  multi dimensional arrays.
+  multi dimensional arrays.

sql/src/main/java/io/crate/analyze/AlterTableAddColumnAnalyzer.java

@@ -31,10 +31,14 @@
 import io.crate.metadata.table.Operation;
 import io.crate.metadata.table.TableInfo;
 import io.crate.sql.tree.AlterTableAddColumn;
-import io.crate.types.CollectionType;
 
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Locale;
+import java.util.Objects;
 
 class AlterTableAddColumnAnalyzer {
 
@@ -102,23 +106,42 @@ private static void ensureColumnLeafsAreNew(AnalyzedColumnDefinition column, Tab
     }
 
     private static void addExistingPrimaryKeys(DocTableInfo tableInfo, AnalyzedTableElements tableElements) {
+        LinkedHashSet<ColumnIdent> pkIncludingAncestors = new LinkedHashSet<>();
         for (ColumnIdent pkIdent : tableInfo.primaryKey()) {
-            if (pkIdent.name().equals("_id")) {
+            if (pkIdent.name().equals(DocSysColumns.Names.ID)) {
                 continue;
             }
-            Reference pkInfo = tableInfo.getReference(pkIdent);
-            assert pkInfo != null : "pk must not be null";
-
-            AnalyzedColumnDefinition pkColumn = new AnalyzedColumnDefinition(null, null);
-            pkColumn.ident(pkIdent);
-            pkColumn.name(pkIdent.name());
-            pkColumn.setPrimaryKeyConstraint();
-
-            assert !(pkInfo.valueType() instanceof CollectionType) : "pk can't be an array";
-            pkColumn.dataType(pkInfo.valueType().getName());
-            tableElements.add(pkColumn);
+            ColumnIdent maybeParent = pkIdent;
+            pkIncludingAncestors.add(maybeParent);
+            while ((maybeParent = maybeParent.getParent()) != null) {
+                pkIncludingAncestors.add(maybeParent);
+            }
+        }
+        ArrayList<ColumnIdent> columnsToBuildHierarchy = new ArrayList<>(pkIncludingAncestors);
+        // We want to have the root columns earlier in the list so that the loop below can be sure parent elements are already present in `columns`
+        columnsToBuildHierarchy.sort(Comparator.comparingInt(c -> c.path().size()));
+        HashMap<ColumnIdent, AnalyzedColumnDefinition> columns = new HashMap<>();
+        for (ColumnIdent column : columnsToBuildHierarchy) {
+            ColumnIdent parent = column.getParent();
+            // sort of `columnsToBuildHierarchy` ensures parent would already have been processed and must be present in columns
+            AnalyzedColumnDefinition parentDef = columns.get(parent);
+            AnalyzedColumnDefinition columnDef = new AnalyzedColumnDefinition(null, parentDef);
+            columns.put(column, columnDef);
+            columnDef.ident(column);
+            if (tableInfo.primaryKey().contains(column)) {
+                columnDef.setPrimaryKeyConstraint();
+            }
+            Reference reference = Objects.requireNonNull(
+                tableInfo.getReference(column),
+                "Must be able to retrieve Reference for any column that is part of `primaryKey()`");
+            columnDef.dataType(reference.valueType().getName());
+            if (parentDef != null) {
+                parentDef.addChild(columnDef);
+            }
+            if (column.isTopLevel()) {
+                tableElements.add(columnDef);
+            }
         }
-
         for (ColumnIdent columnIdent : tableInfo.partitionedBy()) {
             tableElements.changeToPartitionedByColumn(columnIdent, true, tableInfo.ident());
         }

sql/src/main/java/io/crate/analyze/AnalyzedColumnDefinition.java

@@ -377,6 +377,7 @@ void addCopyTo(Set<String> targets) {
     public void ident(ColumnIdent ident) {
         assert this.ident == null : "ident must be null";
         this.ident = ident;
+        this.name = ident.leafName();
     }
 
     boolean isArrayOrInArray() {

sql/src/main/java/io/crate/analyze/AnalyzedTableElements.java

@@ -520,4 +520,5 @@ public List<AnalyzedColumnDefinition> columns() {
     boolean hasGeneratedColumns() {
         return numGeneratedColumns > 0;
     }
+
 }

sql/src/main/java/io/crate/metadata/ColumnIdent.java

@@ -404,4 +404,12 @@ public ColumnIdent prepend(String name) {
         newPath.add(0, this.name);
         return new ColumnIdent(name, newPath);
     }
+
+    public String leafName() {
+        if (path.isEmpty()) {
+            return name;
+        } else {
+            return path.get(path.size() - 1);
+        }
+    }
 }

sql/src/test/java/io/crate/analyze/AlterTableAddColumnAnalyzerTest.java

@@ -21,6 +21,7 @@
 
 package io.crate.analyze;
 
+import com.carrotsearch.randomizedtesting.annotations.Repeat;
 import io.crate.common.collections.Maps;
 import io.crate.exceptions.OperationOnInaccessibleRelationException;
 import io.crate.metadata.ColumnIdent;
@@ -51,7 +52,26 @@ public class AlterTableAddColumnAnalyzerTest extends CrateDummyClusterServiceUni
 
     @Before
     public void prepare() throws IOException {
-        e = SQLExecutor.builder(clusterService).enableDefaultTables().build();
+        e = SQLExecutor.builder(clusterService)
+            .enableDefaultTables()
+            .addTable("create table nested_pks (" +
+                      "     pk object as (a int, b object as (c int))," +
+                      "     primary key (pk['a'], pk['b']['c'])" +
+                      ")")
+            .build();
+    }
+
+
+    @Test
+    public void test_can_add_column_to_table_with_multiple_nested_pks() {
+        AddColumnAnalyzedStatement analysis = e.analyze("alter table nested_pks add x int");
+        assertThat(
+            analysis.analyzedTableElements().toMapping().toString(),
+            is("{_meta={primary_keys=[pk.a, pk.b.c]}, " +
+               "properties={" +
+                    "x={position=2, type=integer}, " +
+                    "pk={dynamic=true, type=object, properties={a={type=integer}, b={dynamic=true, type=object, properties={c={type=integer}}}}}}}")
+        );
     }
 
     @Test