@@ -986,93 +986,35 @@ check_functional_grouping(Oid relid,
986
986
List * grouping_columns ,
987
987
List * * constraintDeps )
988
988
{
989
- bool result = false;
990
- Relation pg_constraint ;
991
- HeapTuple tuple ;
992
- SysScanDesc scan ;
993
- ScanKeyData skey [1 ];
994
-
995
- /* Scan pg_constraint for constraints of the target rel */
996
- pg_constraint = heap_open (ConstraintRelationId , AccessShareLock );
997
-
998
- ScanKeyInit (& skey [0 ],
999
- Anum_pg_constraint_conrelid ,
1000
- BTEqualStrategyNumber , F_OIDEQ ,
1001
- ObjectIdGetDatum (relid ));
1002
-
1003
- scan = systable_beginscan (pg_constraint , ConstraintRelidIndexId , true,
1004
- NULL , 1 , skey );
1005
-
1006
- while (HeapTupleIsValid (tuple = systable_getnext (scan )))
989
+ Bitmapset * pkattnos ;
990
+ Bitmapset * groupbyattnos ;
991
+ Oid constraintOid ;
992
+ ListCell * gl ;
993
+
994
+ /* If the rel has no PK, then we can't prove functional dependency */
995
+ pkattnos = get_primary_key_attnos (relid , false, & constraintOid );
996
+ if (pkattnos == NULL )
997
+ return false;
998
+
999
+ /* Identify all the rel's columns that appear in grouping_columns */
1000
+ groupbyattnos = NULL ;
1001
+ foreach (gl , grouping_columns )
1007
1002
{
1008
- Form_pg_constraint con = (Form_pg_constraint ) GETSTRUCT (tuple );
1009
- Datum adatum ;
1010
- bool isNull ;
1011
- ArrayType * arr ;
1012
- int16 * attnums ;
1013
- int numkeys ;
1014
- int i ;
1015
- bool found_col ;
1003
+ Var * gvar = (Var * ) lfirst (gl );
1016
1004
1017
- /* Only PK constraints are of interest for now, see comment above */
1018
- if (con -> contype != CONSTRAINT_PRIMARY )
1019
- continue ;
1020
- /* Constraint must be non-deferrable */
1021
- if (con -> condeferrable )
1022
- continue ;
1023
-
1024
- /* Extract the conkey array, ie, attnums of PK's columns */
1025
- adatum = heap_getattr (tuple , Anum_pg_constraint_conkey ,
1026
- RelationGetDescr (pg_constraint ), & isNull );
1027
- if (isNull )
1028
- elog (ERROR , "null conkey for constraint %u" ,
1029
- HeapTupleGetOid (tuple ));
1030
- arr = DatumGetArrayTypeP (adatum ); /* ensure not toasted */
1031
- numkeys = ARR_DIMS (arr )[0 ];
1032
- if (ARR_NDIM (arr ) != 1 ||
1033
- numkeys < 0 ||
1034
- ARR_HASNULL (arr ) ||
1035
- ARR_ELEMTYPE (arr ) != INT2OID )
1036
- elog (ERROR , "conkey is not a 1-D smallint array" );
1037
- attnums = (int16 * ) ARR_DATA_PTR (arr );
1038
-
1039
- found_col = false;
1040
- for (i = 0 ; i < numkeys ; i ++ )
1041
- {
1042
- AttrNumber attnum = attnums [i ];
1043
- ListCell * gl ;
1044
-
1045
- found_col = false;
1046
- foreach (gl , grouping_columns )
1047
- {
1048
- Var * gvar = (Var * ) lfirst (gl );
1049
-
1050
- if (IsA (gvar , Var ) &&
1051
- gvar -> varno == varno &&
1052
- gvar -> varlevelsup == varlevelsup &&
1053
- gvar -> varattno == attnum )
1054
- {
1055
- found_col = true;
1056
- break ;
1057
- }
1058
- }
1059
- if (!found_col )
1060
- break ;
1061
- }
1062
-
1063
- if (found_col )
1064
- {
1065
- /* The PK is a subset of grouping_columns, so we win */
1066
- * constraintDeps = lappend_oid (* constraintDeps ,
1067
- HeapTupleGetOid (tuple ));
1068
- result = true;
1069
- break ;
1070
- }
1005
+ if (IsA (gvar , Var ) &&
1006
+ gvar -> varno == varno &&
1007
+ gvar -> varlevelsup == varlevelsup )
1008
+ groupbyattnos = bms_add_member (groupbyattnos ,
1009
+ gvar -> varattno - FirstLowInvalidHeapAttributeNumber );
1071
1010
}
1072
1011
1073
- systable_endscan (scan );
1074
-
1075
- heap_close (pg_constraint , AccessShareLock );
1012
+ if (bms_is_subset (pkattnos , groupbyattnos ))
1013
+ {
1014
+ /* The PK is a subset of grouping_columns, so we win */
1015
+ * constraintDeps = lappend_oid (* constraintDeps , constraintOid );
1016
+ return true;
1017
+ }
1076
1018
1077
- return result ;
1019
+ return false ;
1078
1020
}
0 commit comments