parser: Add user defined column families to CTAS.

docs/generated/sql/bnf/stmt_block.bnf

@@ -1694,7 +1694,7 @@ partition_by ::=
 	| 'PARTITION' 'BY' 'NOTHING'
 
 create_as_table_defs ::=
-	( column_name create_as_col_qual_list ) ( ( ',' create_as_constraint_def | ',' column_name create_as_col_qual_list ) )*
+	( column_name create_as_col_qual_list ) ( ( ',' column_name create_as_col_qual_list | ',' family_def | ',' create_as_constraint_def ) )*
 
 common_table_expr ::=
 	table_alias_name opt_column_list 'AS' '(' preparable_stmt ')'
@@ -2136,6 +2136,7 @@ range_partition ::=
 
 create_as_col_qualification ::=
 	create_as_col_qualification_elem
+	| 'FAMILY' family_name
 
 create_as_constraint_elem ::=
 	'PRIMARY' 'KEY' '(' create_as_params ')'

pkg/sql/logictest/testdata/logic_test/create_as

@@ -236,3 +236,65 @@ foo11  CREATE TABLE foo11 (
 statement error pq: multiple primary keys for table "foo12" are not allowed
 CREATE TABLE foo12 (x PRIMARY KEY, y, PRIMARY KEY(y)) AS VALUES (1, 2), (3, 4);
 
+# Check that CREATE TABLE AS allows users to specify column families.
+statement ok
+CREATE TABLE abcd(
+  a INT PRIMARY KEY,
+  b INT,
+  c INT,
+  d INT
+);
+
+# Test column qualifiers to define column families.
+statement ok
+CREATE TABLE foo12 (a PRIMARY KEY FAMILY f1, b, c FAMILY fam_1_c, d) AS SELECT * FROM abcd;
+
+query TT
+SHOW CREATE TABLE foo12
+----
+foo12  CREATE TABLE foo12 (
+      a INT8 NOT NULL,
+      b INT8 NULL,
+      c INT8 NULL,
+      d INT8 NULL,
+      CONSTRAINT "primary" PRIMARY KEY (a ASC),
+      FAMILY f1 (a, b, d),
+      FAMILY fam_1_c (c)
+)
+
+# Test constraint style definition of column families.
+statement ok
+CREATE TABLE foo13 (a, b, c, d, PRIMARY KEY(a, b), FAMILY pk (a, b), FAMILY (c, d)) AS SELECT * FROM abcd;
+
+query TT
+SHOW CREATE TABLE foo13
+----
+foo13  CREATE TABLE foo13 (
+      a INT8 NOT NULL,
+      b INT8 NOT NULL,
+      c INT8 NULL,
+      d INT8 NULL,
+      CONSTRAINT "primary" PRIMARY KEY (a ASC, b ASC),
+      FAMILY pk (a, b),
+      FAMILY fam_1_c_d (c, d)
+)
+
+# Test renaming columns still preserves the column families.
+statement ok
+ALTER TABLE foo13 RENAME d TO z
+
+statement ok
+ALTER TABLE foo13 RENAME c TO e
+
+query TT
+SHOW CREATE TABLE foo13
+----
+foo13  CREATE TABLE foo13 (
+            a INT8 NOT NULL,
+            b INT8 NOT NULL,
+            e INT8 NULL,
+            z INT8 NULL,
+            CONSTRAINT "primary" PRIMARY KEY (a ASC, b ASC),
+            FAMILY pk (a, b),
+            FAMILY fam_1_c_d (e, z)
+)

pkg/sql/parser/parse_test.go

@@ -257,6 +257,11 @@ func TestParse(t *testing.T) {
 		{`CREATE TABLE IF NOT EXISTS a (z PRIMARY KEY) AS SELECT * FROM b`},
 		{`CREATE TABLE a (x, y, z, PRIMARY KEY (x, y, z)) AS SELECT * FROM b`},
 		{`CREATE TABLE IF NOT EXISTS a (x, y, z, PRIMARY KEY (x, y, z)) AS SELECT * FROM b`},
+		{`CREATE TABLE a (x, FAMILY (x)) AS SELECT * FROM b`},
+		{`CREATE TABLE IF NOT EXISTS a (x, FAMILY (x)) AS SELECT * FROM b`},
+		{`CREATE TABLE a (x, y FAMILY f1) AS SELECT * FROM b`},
+		{`CREATE TABLE IF NOT EXISTS a (x, y FAMILY f1) AS SELECT * FROM b`},
+
 		{`CREATE TABLE a (b STRING COLLATE de)`},
 		{`CREATE TABLE a (b STRING(3) COLLATE de)`},
 		{`CREATE TABLE a (b STRING[] COLLATE de)`},

pkg/sql/parser/sql.y

@@ -4443,11 +4443,6 @@ create_as_table_defs:
     var colToTableDef tree.TableDef = tableDef
     $$.val = tree.TableDefs{colToTableDef}
   }
-| create_as_table_defs ',' create_as_constraint_def
-  {
-    var constraintToTableDef tree.TableDef = $3.constraintDef()
-    $$.val = append($1.tblDefs(), constraintToTableDef)
-  }
 | create_as_table_defs ',' column_name create_as_col_qual_list
   {
     tableDef, err := tree.NewColumnTableDef(tree.Name($3), nil, false, $4.colQuals())
@@ -4459,6 +4454,15 @@ create_as_table_defs:
 
     $$.val = append($1.tblDefs(), colToTableDef)
   }
+| create_as_table_defs ',' family_def
+  {
+    $$.val = append($1.tblDefs(), $3.tblDef())
+  }
+| create_as_table_defs ',' create_as_constraint_def
+{
+  var constraintToTableDef tree.TableDef = $3.constraintDef()
+  $$.val = append($1.tblDefs(), constraintToTableDef)
+}
 
 create_as_constraint_def:
   create_as_constraint_elem
@@ -4508,6 +4512,10 @@ create_as_col_qualification:
   {
     $$.val = tree.NamedColumnQualification{Qualification: $1.colQualElem()}
   }
+| FAMILY family_name
+  {
+    $$.val = tree.NamedColumnQualification{Qualification: &tree.ColumnFamilyConstraint{Family: tree.Name($2)}}
+  }
 
 create_as_col_qualification_elem:
   PRIMARY KEY