Fix duplicate distributed keys for CTAS

To keep it consistent with the "Create table" syntax, CTAS should also
disallow duplicate distributed keys, otherwise backup and restore will
mess up.

src/backend/cdb/cdbmutate.c

@@ -351,9 +351,27 @@ apply_motion(PlannerInfo *root, Plan *plan, Query *query)
 					
 									if (equal(var1,new_var))
 									{
+										int i;
+
 										/* If it is, use it to partition the result table, to avoid
 										 * unnecessary redistibution of data */
 										Assert(targetPolicy->nattrs < MaxPolicyAttributeNumber);
+
+										/* check duplicate distribute key */
+										for (i = 0; i < targetPolicy->nattrs; i++)
+										{
+											if (targetPolicy->attrs[i] != n)
+												continue;
+
+											TargetEntry *target = get_tle_by_resno(plan->targetlist, n);
+
+											ereport(ERROR,
+													(errcode(ERRCODE_DUPLICATE_COLUMN),
+													 errmsg("duplicate DISTRIBUTED BY column '%s'",
+															target->resname ? target->resname : "???")));
+
+										}
+
 										targetPolicy->attrs[targetPolicy->nattrs++] = n;
 										found_expr = true;
 										break;

src/test/regress/expected/qp_misc_jiras.out

@@ -4595,6 +4595,12 @@ select test();
  
 (1 row)
 
+-- Test duplicate distribute keys
+CREATE TABLE qp_misc_jiras.ctas_dup_dk_src (col1 int, col2 int);
+CREATE TABLE qp_misc_jiras.ctas_dup_dk as SELECT distinct col2 as c1, col2 as c2 from qp_misc_jiras.ctas_dup_dk_src;
+ERROR:  duplicate DISTRIBUTED BY column 'c1'
+SELECT distinct col2 c1, col2 c2 into qp_misc_jiras.ctas_dup_dk_1 from qp_misc_jiras.ctas_dup_dk_src;
+ERROR:  duplicate DISTRIBUTED BY column 'c1'
 -- start_ignore
 drop schema qp_misc_jiras cascade;
 NOTICE:  drop cascades to table qp_misc_jiras._tbl10050_test

src/test/regress/expected/qp_misc_jiras_optimizer.out

@@ -4624,6 +4624,10 @@ select test();
  
 (1 row)
 
+-- Test duplicate distribute keys
+CREATE TABLE qp_misc_jiras.ctas_dup_dk_src (col1 int, col2 int);
+CREATE TABLE qp_misc_jiras.ctas_dup_dk as SELECT distinct col2 as c1, col2 as c2 from qp_misc_jiras.ctas_dup_dk_src;
+SELECT distinct col2 c1, col2 c2 into qp_misc_jiras.ctas_dup_dk_1 from qp_misc_jiras.ctas_dup_dk_src;
 -- start_ignore
 drop schema qp_misc_jiras cascade;
 NOTICE:  drop cascades to table qp_misc_jiras._tbl10050_test

src/test/regress/sql/qp_misc_jiras.sql

@@ -2590,6 +2590,12 @@ select test();
 select test();
 select test();
 select test();
+
+-- Test duplicate distribute keys
+CREATE TABLE qp_misc_jiras.ctas_dup_dk_src (col1 int, col2 int);
+CREATE TABLE qp_misc_jiras.ctas_dup_dk as SELECT distinct col2 as c1, col2 as c2 from qp_misc_jiras.ctas_dup_dk_src;
+SELECT distinct col2 c1, col2 c2 into qp_misc_jiras.ctas_dup_dk_1 from qp_misc_jiras.ctas_dup_dk_src;
+
 -- start_ignore
 drop schema qp_misc_jiras cascade;
 -- end_ignore