Repository: vitessio/vitess
Pull request #10128 (2022-04-22 12:28:45)

feat gen4: allow pushing aggregations inside derived tables
## Description
Adds the capability of building aggregation plans on top of derived tables.

## Checklist
-   [ ] "Backport me!" label has been added if this change should be back ported
-   [x] Tests were added or are not required
-   [x] Documentation was added or is not required



Patch:
diff --git a/go/vt/vtgate/planbuilder/aggregation_pushing.go b/go/vt/vtgate/planbuilder/aggregation_pushing.go
index 75997117f2..4707c667b6 100644
--- a/go/vt/vtgate/planbuilder/aggregation_pushing.go
+++ b/go/vt/vtgate/planbuilder/aggregation_pushing.go
@@ -34,23 +34,30 @@ func (hp *horizonPlanning) pushAggregation(
 	grouping []abstract.GroupBy,
 	aggregations []abstract.Aggr,
 	ignoreOutputOrder bool,
-) (groupingOffsets []offsets, outputAggrsOffset [][]offsets, err error) {
+) (output logicalPlan, groupingOffsets []offsets, outputAggrsOffset [][]offsets, err error) {
 	switch plan := plan.(type) {
 	case *routeGen4:
+		output = plan
 		groupingOffsets, outputAggrsOffset, err = pushAggrOnRoute(ctx, plan, aggregations, grouping, ignoreOutputOrder)
-		if err != nil {
-			return nil, nil, err
-		}
 		return
 
 	case *joinGen4:
-		return hp.pushAggrOnJoin(ctx, grouping, aggregations, plan)
+		output = plan
+		groupingOffsets, outputAggrsOffset, err = hp.pushAggrOnJoin(ctx, plan, grouping, aggregations)
+		return
 
 	case *semiJoin:
-		return hp.pushAggrOnSemiJoin(ctx, grouping, aggregations, plan, ignoreOutputOrder)
+		output = plan
+		groupingOffsets, outputAggrsOffset, err = hp.pushAggrOnSemiJoin(ctx, plan, grouping, aggregations, ignoreOutputOrder)
+		return
+
+	case *simpleProjection:
+		// we just remove the simpleProjection. We are doing an OA on top anyway, so no need to clean up the output columns
+		return hp.pushAggregation(ctx, plan.input, grouping, aggregations, ignoreOutputOrder)
 
 	default:
-		return nil, nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "using aggregation on top of a %T plan is not yet supported", plan)
+		err = vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "using aggregation on top of a %T plan is not yet supported", plan)
+		return
 	}
 }
 
@@ -195,9 +202,9 @@ vtgate level, we can offload most of the work to MySQL, and at the vtgate just s
 */
 func (hp *horizonPlanning) pushAggrOnJoin(
 	ctx *plancontext.PlanningContext,
+	join *joinGen4,
 	grouping []abstract.GroupBy,
 	aggregations []abstract.Aggr,
-	join *joinGen4,
 ) ([]offsets, [][]offsets, error) {
 	// First we separate aggregations according to which side the dependencies are coming from
 	lhsAggrs, rhsAggrs, err := splitAggregationsToLeftAndRight(ctx, aggregations, join)
@@ -220,15 +227,16 @@ func (hp *horizonPlanning) pushAggrOnJoin(
 	}
 
 	// Next we push the aggregations to both sides
-	lhsOffsets, lhsAggrOffsets, err := hp.filteredPushAggregation(ctx, join.Left, lhsGrouping, lhsAggrs, true)
+	newLHS, lhsOffsets, lhsAggrOffsets, err := hp.filteredPushAggregation(ctx, join.Left, lhsGrouping, lhsAggrs, true)
 	if err != nil {
 		return nil, nil, err
 	}
 
-	rhsOffsets, rhsAggrOffsets, err := hp.filteredPushAggregation(ctx, join.Right, rhsGrouping, rhsAggrs, true)
+	newRHS, rhsOffsets, rhsAggrOffsets, err := hp.filteredPushAggregation(ctx, join.Right, rhsGrouping, rhsAggrs, true)
 	if err != nil {
 		return nil, nil, err
 	}
+	join.Left, join.Right = newLHS, newRHS
 
 	// Next, we have to pass through the grouping values through the join and the projection we add on top
 	// We added new groupings to the LHS because of the join condition, so we don't want to pass through everything,
@@ -283,9 +291,9 @@ That way we get the aggregation grouped by the column we need to use to decide i
 */
 func (hp *horizonPlanning) pushAggrOnSemiJoin(
 	ctx *plancontext.PlanningContext,
+	join *semiJoin,
 	grouping []abstract.GroupBy,
 	aggregations []abstract.Aggr,
-	join *semiJoin,
 	ignoreOutputOrder bool,
 ) ([]offsets, [][]offsets, error) {
 	// We need to group by the columns used in the join condition.
@@ -296,10 +304,11 @@ func (hp *horizonPlanning) pushAggrOnSemiJoin(
 	}
 
 	totalGrouping := append(grouping, lhsCols...)
-	groupingOffsets, aggrParams, err := hp.pushAggregation(ctx, join.lhs, totalGrouping, aggregations, ignoreOutputOrder)
+	newLeft, groupingOffsets, aggrParams, err := hp.pushAggregation(ctx, join.lhs, totalGrouping, aggregations, ignoreOutputOrder)
 	if err != nil {
 		return nil, nil, err
 	}
+	join.lhs = newLeft
 
 	outputGroupings := make([]offsets, 0, len(grouping))
 	for idx := range grouping {
@@ -321,7 +330,7 @@ func (hp *horizonPlanning) filteredPushAggregation(
 	grouping []abstract.GroupBy,
 	aggregations []*abstract.Aggr,
 	ignoreOutputOrder bool,
-) (groupingOffsets []offsets, outputAggrs [][]offsets, err error) {
+) (out logicalPlan, groupingOffsets []offsets, outputAggrs [][]offsets, err error) {
 	used := make([]bool, len(aggregations))
 	var aggrs []abstract.Aggr
 
@@ -331,9 +340,9 @@ func (hp *horizonPlanning) filteredPushAggregation(
 			aggrs = append(aggrs, *aggr)
 		}
 	}
-	groupingOffsets, pushedAggrs, err := hp.pushAggregation(ctx, plan, grouping, aggrs, ignoreOutputOrder)
+	newplan, groupingOffsets, pushedAggrs, err := hp.pushAggregation(ctx, plan, grouping, aggrs, ignoreOutputOrder)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
 	}
 	idx := 0
 	for _, b := range used {
@@ -344,7 +353,7 @@ func (hp *horizonPlanning) filteredPushAggregation(
 		outputAggrs = append(outputAggrs, pushedAggrs[idx])
 		idx++
 	}
-	return groupingOffsets, outputAggrs, nil
+	return newplan, groupingOffsets, outputAggrs, nil
 }
 
 func isMinOrMax(in engine.AggregateOpcode) bool {
diff --git a/go/vt/vtgate/planbuilder/horizon_planning.go b/go/vt/vtgate/planbuilder/horizon_planning.go
index f31ed95048..7831c13994 100644
--- a/go/vt/vtgate/planbuilder/horizon_planning.go
+++ b/go/vt/vtgate/planbuilder/horizon_planning.go
@@ -551,11 +551,13 @@ func (hp *horizonPlanning) planAggrUsingOA(
 		oa.preProcess = true
 	}
 
-	groupingOffsets, aggrParamOffsets, err := hp.pushAggregation(ctx, plan, grouping, aggrs, false)
+	newPlan, groupingOffsets, aggrParamOffsets, err := hp.pushAggregation(ctx, plan, grouping, aggrs, false)
 	if err != nil {
 		return nil, err
 	}
 
+	plan = newPlan
+
 	_, isRoute := plan.(*routeGen4)
 	needsProj := !isRoute
 	var aggPlan = plan


Test patch:
diff --git a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
index 0e342f6d61..3810378a82 100644
--- a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
@@ -3895,3 +3895,55 @@ Gen4 plan same as above
 "select count(distinct user_id, name) from user"
 "unsupported: only one expression allowed inside aggregates: count(distinct user_id, `name`)"
 Gen4 error: aggregate functions take a single argument 'count(distinct user_id, `name`)'
+
+"select sum(col) from (select user.col as col, 32 from user join user_extra) t"
+"unsupported: cross-shard query with aggregates"
+{
+  "QueryType": "SELECT",
+  "Original": "select sum(col) from (select user.col as col, 32 from user join user_extra) t",
+  "Instructions": {
+    "OperatorType": "Aggregate",
+    "Variant": "Scalar",
+    "Aggregates": "sum(0) AS sum(col)",
+    "Inputs": [
+      {
+        "OperatorType": "Projection",
+        "Expressions": [
+          "[COLUMN 2] * [COLUMN 3] as sum(col)"
+        ],
+        "Inputs": [
+          {
+            "OperatorType": "Join",
+            "Variant": "Join",
+            "JoinColumnIndexes": "L:0,L:1,L:2,R:0",
+            "TableName": "`user`_user_extra",
+            "Inputs": [
+              {
+                "OperatorType": "Route",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "user",
+                  "Sharded": true
+                },
+                "FieldQuery": "select `user`.col as col, 32, sum(col) from `user` where 1 != 1",
+                "Query": "select `user`.col as col, 32, sum(col) from `user`",
+                "Table": "`user`"
+              },
+              {
+                "OperatorType": "Route",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "user",
+                  "Sharded": true
+                },
+                "FieldQuery": "select count(*) from user_extra where 1 != 1",
+                "Query": "select count(*) from user_extra",
+                "Table": "user_extra"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  }
+}
diff --git a/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt b/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt
index e8ec3383e2..74b962f20a 100644
--- a/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt
@@ -418,7 +418,182 @@ Gen4 error: unsupported: cross-shard correlated subquery
 # TPC-H query 7
 "select supp_nation, cust_nation, l_year, sum(volume) as revenue from (select n1.n_name as supp_nation, n2.n_name as cust_nation, extract(year from l_shipdate) as l_year, l_extendedprice * (1 - l_discount) as volume from supplier, lineitem, orders, customer, nation n1, nation n2 where s_suppkey = l_suppkey and o_orderkey = l_orderkey and c_custkey = o_custkey and s_nationkey = n1.n_nationkey and c_nationkey = n2.n_nationkey and ((n1.n_name = 'FRANCE' and n2.n_name = 'GERMANY') or (n1.n_name = 'GERMANY' and n2.n_name = 'FRANCE')) and l_shipdate between date('1995-01-01') and date('1996-12-31')) as shipping group by supp_nation, cust_nation, l_year order by supp_nation, cust_nation, l_year"
 "unsupported: cross-shard query with aggregates"
-Gen4 error: using aggregation on top of a *planbuilder.simpleProjection plan is not yet supported
+{
+  "QueryType": "SELECT",
+  "Original": "select supp_nation, cust_nation, l_year, sum(volume) as revenue from (select n1.n_name as supp_nation, n2.n_name as cust_nation, extract(year from l_shipdate) as l_year, l_extendedprice * (1 - l_discount) as volume from supplier, lineitem, orders, customer, nation n1, nation n2 where s_suppkey = l_suppkey and o_orderkey = l_orderkey and c_custkey = o_custkey and s_nationkey = n1.n_nationkey and c_nationkey = n2.n_nationkey and ((n1.n_name = 'FRANCE' and n2.n_name = 'GERMANY') or (n1.n_name = 'GERMANY' and n2.n_name = 'FRANCE')) and l_shipdate between date('1995-01-01') and date('1996-12-31')) as shipping group by supp_nation, cust_nation, l_year order by supp_nation, cust_nation, l_year",
+  "Instructions": {
+    "OperatorType": "Aggregate",
+    "Variant": "Ordered",
+    "Aggregates": "sum(3) AS revenue",
+    "GroupBy": "(0|6), (1|5), (2|4)",
+    "ResultColumns": 4,
+    "Inputs": [
+      {
+        "OperatorType": "Projection",
+        "Expressions": [
+          "[COLUMN 4]",
+          "[COLUMN 5]",
+          "[COLUMN 6]",
+          "(((([COLUMN 10] * [COLUMN 11]) * [COLUMN 12]) * [COLUMN 13]) * [COLUMN 14]) * [COLUMN 15] as revenue",
+          "[COLUMN 9]",
+          "[COLUMN 8]",
+          "[COLUMN 7]"
+        ],
+        "Inputs": [
+          {
+            "OperatorType": "Sort",
+            "Variant": "Memory",
+            "OrderBy": "(0|16) ASC, (1|17) ASC, (2|18) ASC",
+            "Inputs": [
+              {
+                "OperatorType": "Join",
+                "Variant": "Join",
+                "JoinColumnIndexes": "L:3,R:0,L:4,L:5,L:9,R:1,L:10,L:14,R:2,L:15,L:16,L:17,L:18,L:19,R:3,R:4,L:20,R:5,L:21",
+                "JoinVars": {
+                  "n1_n_name": 2,
+                  "o_custkey": 0
+                },
+                "TableName": "lineitem_orders_supplier_nation_customer_nation",
+                "Inputs": [
+                  {
+                    "OperatorType": "Join",
+                    "Variant": "Join",
+                    "JoinColumnIndexes": "L:1,R:0,R:1,R:2,L:2,L:3,L:5,R:3,R:4,R:5,L:6,L:8,R:6,R:7,R:8,L:9,L:10,L:11,R:9,R:10,R:11,L:12",
+                    "JoinVars": {
+                      "l_suppkey": 0
+                    },
+                    "TableName": "lineitem_orders_supplier_nation",
+                    "Inputs": [
+                      {
+                        "OperatorType": "Join",
+                        "Variant": "Join",
+                        "JoinColumnIndexes": "L:1,R:0,L:2,L:3,L:1,R:0,L:2,L:6,R:2,L:7,L:4,R:1,L:8",
+                        "JoinVars": {
+                          "l_orderkey": 0
+                        },
+                        "TableName": "lineitem_orders",
+                        "Inputs": [
+                          {
+                            "OperatorType": "Route",
+                            "Variant": "Scatter",
+                            "Keyspace": {
+                              "Name": "main",
+                              "Sharded": true
+                            },
+                            "FieldQuery": "select l_orderkey, l_suppkey, extract(year from l_shipdate) as l_year, l_extendedprice * (1 - l_discount) as volume, sum(volume) as revenue, weight_string(l_orderkey), weight_string(l_suppkey), weight_string(extract(year from l_shipdate)), weight_string(extract(year from l_shipdate)) from lineitem where 1 != 1 group by l_orderkey, weight_string(l_orderkey), l_suppkey, weight_string(l_suppkey), l_year, weight_string(l_year)",
+                            "Query": "select l_orderkey, l_suppkey, extract(year from l_shipdate) as l_year, l_extendedprice * (1 - l_discount) as volume, sum(volume) as revenue, weight_string(l_orderkey), weight_string(l_suppkey), weight_string(extract(year from l_shipdate)), weight_string(extract(year from l_shipdate)) from lineitem where l_shipdate between date('1995-01-01') and date('1996-12-31') group by l_orderkey, weight_string(l_orderkey), l_suppkey, weight_string(l_suppkey), l_year, weight_string(l_year)",
+                            "Table": "lineitem"
+                          },
+                          {
+                            "OperatorType": "Route",
+                            "Variant": "EqualUnique",
+                            "Keyspace": {
+                              "Name": "main",
+                              "Sharded": true
+                            },
+                            "FieldQuery": "select o_custkey, count(*), weight_string(o_custkey) from orders where 1 != 1 group by o_custkey, weight_string(o_custkey)",
+                            "Query": "select o_custkey, count(*), weight_string(o_custkey) from orders where o_orderkey = :l_orderkey group by o_custkey, weight_string(o_custkey)",
+                            "Table": "orders",
+                            "Values": [
+                              ":l_orderkey"
+                            ],
+                            "Vindex": "hash"
+                          }
+                        ]
+                      },
+                      {
+                        "OperatorType": "Join",
+                        "Variant": "Join",
+                        "JoinColumnIndexes": "R:0,R:0,R:1,R:0,R:0,R:1,R:3,R:3,R:4,L:1,R:2,R:5",
+                        "JoinVars": {
+                          "s_nationkey": 0
+                        },
+                        "TableName": "supplier_nation",
+                        "Inputs": [
+                          {
+                            "OperatorType": "Route",
+                            "Variant": "EqualUnique",
+                            "Keyspace": {
+                              "Name": "main",
+                              "Sharded": true
+                            },
+                            "FieldQuery": "select s_nationkey, count(*), weight_string(s_nationkey) from supplier where 1 != 1 group by s_nationkey, weight_string(s_nationkey)",
+                            "Query": "select s_nationkey, count(*), weight_string(s_nationkey) from supplier where s_suppkey = :l_suppkey group by s_nationkey, weight_string(s_nationkey)",
+                            "Table": "supplier",
+                            "Values": [
+                              ":l_suppkey"
+                            ],
+                            "Vindex": "hash"
+                          },
+                          {
+                            "OperatorType": "Route",
+                            "Variant": "EqualUnique",
+                            "Keyspace": {
+                              "Name": "main",
+                              "Sharded": true
+                            },
+                            "FieldQuery": "select n1.n_name, n1.n_name as supp_nation, count(*), weight_string(n1.n_name), weight_string(n1.n_name), weight_string(n1.n_name) from nation as n1 where 1 != 1 group by n1.n_name, weight_string(n1.n_name), supp_nation, weight_string(supp_nation)",
+                            "Query": "select n1.n_name, n1.n_name as supp_nation, count(*), weight_string(n1.n_name), weight_string(n1.n_name), weight_string(n1.n_name) from nation as n1 where n1.n_nationkey = :s_nationkey group by n1.n_name, weight_string(n1.n_name), supp_nation, weight_string(supp_nation)",
+                            "Table": "nation",
+                            "Values": [
+                              ":s_nationkey"
+                            ],
+                            "Vindex": "hash"
+                          }
+                        ]
+                      }
+                    ]
+                  },
+                  {
+                    "OperatorType": "Join",
+                    "Variant": "Join",
+                    "JoinColumnIndexes": "R:0,R:0,R:2,L:1,R:1,R:3",
+                    "JoinVars": {
+                      "c_nationkey": 0
+                    },
+                    "TableName": "customer_nation",
+                    "Inputs": [
+                      {
+                        "OperatorType": "Route",
+                        "Variant": "EqualUnique",
+                        "Keyspace": {
+                          "Name": "main",
+                          "Sharded": true
+                        },
+                        "FieldQuery": "select c_nationkey, count(*), weight_string(c_nationkey) from customer where 1 != 1 group by c_nationkey, weight_string(c_nationkey)",
+                        "Query": "select c_nationkey, count(*), weight_string(c_nationkey) from customer where c_custkey = :o_custkey group by c_nationkey, weight_string(c_nationkey)",
+                        "Table": "customer",
+                        "Values": [
+                          ":o_custkey"
+                        ],
+                        "Vindex": "hash"
+                      },
+                      {
+                        "OperatorType": "Route",
+                        "Variant": "EqualUnique",
+                        "Keyspace": {
+                          "Name": "main",
+                          "Sharded": true
+                        },
+                        "FieldQuery": "select n2.n_name as cust_nation, count(*), weight_string(n2.n_name), weight_string(n2.n_name) from nation as n2 where 1 != 1 group by cust_nation, weight_string(cust_nation)",
+                        "Query": "select n2.n_name as cust_nation, count(*), weight_string(n2.n_name), weight_string(n2.n_name) from nation as n2 where n2.n_nationkey = :c_nationkey and (:n1_n_name = 'FRANCE' and n2.n_name = 'GERMANY' or :n1_n_name = 'GERMANY' and n2.n_name = 'FRANCE') group by cust_nation, weight_string(cust_nation)",
+                        "Table": "nation",
+                        "Values": [
+                          ":c_nationkey"
+                        ],
+                        "Vindex": "hash"
+                      }
+                    ]
+                  }
+                ]
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  }
+}
 
 # TPC-H query 8
 "select o_year, sum(case when nation = 'BRAZIL' then volume else 0 end) / sum(volume) as mkt_share from ( select extract(year from o_orderdate) as o_year, l_extendedprice * (1 - l_discount) as volume, n2.n_name as nation from part, supplier, lineitem, orders, customer, nation n1, nation n2, region where p_partkey = l_partkey and s_suppkey = l_suppkey and l_orderkey = o_orderkey and o_custkey = c_custkey and c_nationkey = n1.n_nationkey and n1.n_regionkey = r_regionkey and r_name = 'AMERICA' and s_nationkey = n2.n_nationkey and o_orderdate between date '1995-01-01' and date('1996-12-31') and p_type = 'ECONOMY ANODIZED STEEL' ) as all_nations group by o_year order by o_year"
@@ -428,7 +603,7 @@ Gen4 plan same as above
 # TPC-H query 9
 "select nation, o_year, sum(amount) as sum_profit from ( select n_name as nation, extract(year from o_orderdate) as o_year, l_extendedprice * (1 - l_discount) - ps_supplycost * l_quantity as amount from part, supplier, lineitem, partsupp, orders, nation where s_suppkey = l_suppkey and ps_suppkey = l_suppkey and ps_partkey = l_partkey and p_partkey = l_partkey and o_orderkey = l_orderkey and s_nationkey = n_nationkey and p_name like '%green%' ) as profit group by nation, o_year order by nation, o_year desc"
 "unsupported: cross-shard query with aggregates"
-Gen4 error: using aggregation on top of a *planbuilder.simpleProjection plan is not yet supported
+Gen4 error: aggregation on columns from different sources not supported yet
 
 # TPC-H query 10
 "select c_custkey, c_name, sum(l_extendedprice * (1 - l_discount)) as revenue, c_acctbal, n_name, c_address, c_phone, c_comment from customer, orders, lineitem, nation where c_custkey = o_custkey and l_orderkey = o_orderkey and o_orderdate >= date('1993-10-01') and o_orderdate < date('1993-10-01') + interval '3' month and l_returnflag = 'R' and c_nationkey = n_nationkey group by c_custkey, c_name, c_acctbal, c_phone, n_name, c_address, c_comment order by revenue desc limit 20"
@@ -657,7 +832,7 @@ Gen4 error: unsupported: in scatter query: complex aggregate expression
 
 # TPC-H query 13
 "select c_count, count(*) as custdist from ( select c_custkey, count(o_orderkey) from customer left outer join orders on c_custkey = o_custkey and o_comment not like '%special%requests%' group by c_custkey ) as c_orders(c_custkey, c_count) group by c_count order by custdist desc, c_count desc"
-"using aggregation on top of a *planbuilder.simpleProjection plan is not yet supported"
+"using aggregation on top of a *planbuilder.orderedAggregate plan is not yet supported"
 Gen4 plan same as above
 
 # TPC-H query 14
diff --git a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
index 50457d0ef1..ad8903d62a 100644
--- a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
@@ -510,7 +510,7 @@ Gen4 error: unsupported: in scatter query: complex order by expression: a + 1
 # aggregation on union
 "select sum(col) from (select col from user union all select col from unsharded) t"
 "unsupported: cross-shard query with aggregates"
-Gen4 error: using aggregation on top of a *planbuilder.simpleProjection plan is not yet supported
+Gen4 error: using aggregation on top of a *planbuilder.concatenateGen4 plan is not yet supported
 
 # systable union query in derived table with constraint on outside (without star projection)
 "select id from (select id from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'user_extra' union select id from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'music') `kcu` where `id` = 'primary'"