BRIN: be more strict about required support procs

With improperly defined operator classes, it's possible to get a
Postgres crash because we'd try to invoke a procedure that doesn't
exist.  This is because the code is being a bit too trusting that the
opclass is correctly defined.  Add some ereport(ERROR)s for cases where
mandatory support procedures are not defined, transforming the crashes
into errors.

The particular case that was reported is an incomplete opclass in
PostGIS.

Backpatch all the way down to 13.

Reported-by: Tobias Wendorff <tobias.wendorff@tu-dortmund.de>
Diagnosed-by: David Rowley <dgrowleyml@gmail.com>
Reviewed-by: Tomas Vondra <tomas@vondra.me>
Discussion: https://postgr.es/m/fb6d9a35-6c8e-4869-af80-0a4944a793a4@tu-dortmund.de

diff --git a/src/backend/access/brin/brin_bloom.c b/src/backend/access/brin/brin_bloom.c
index c0407e5816b899eea80b8835b77e8d93247e9167..0bcf8e582c8eb0cbe1ca15fd8716b4087ec936d5 100644 (file)
--- a/src/backend/access/brin/brin_bloom.c
+++ b/src/backend/access/brin/brin_bloom.c
@@ -411,7 +411,6 @@ typedef struct BloomOpaque
         * consistency. We may need additional procs in the future.
         */
        FmgrInfo        extra_procinfos[BLOOM_MAX_PROCNUMS];
-       bool            extra_proc_missing[BLOOM_MAX_PROCNUMS];
 } BloomOpaque;
 
 static FmgrInfo *bloom_get_procinfo(BrinDesc *bdesc, uint16 attno,
@@ -684,27 +683,19 @@ bloom_get_procinfo(BrinDesc *bdesc, uint16 attno, uint16 procnum)
         */
        opaque = (BloomOpaque *) bdesc->bd_info[attno - 1]->oi_opaque;
 
-       /*
-        * If we already searched for this proc and didn't find it, don't bother
-        * searching again.
-        */
-       if (opaque->extra_proc_missing[basenum])
-               return NULL;
-
        if (opaque->extra_procinfos[basenum].fn_oid == InvalidOid)
        {
                if (RegProcedureIsValid(index_getprocid(bdesc->bd_index, attno,
                                                                                                procnum)))
-               {
                        fmgr_info_copy(&opaque->extra_procinfos[basenum],
                                                   index_getprocinfo(bdesc->bd_index, attno, procnum),
                                                   bdesc->bd_context);
-               }
                else
-               {
-                       opaque->extra_proc_missing[basenum] = true;
-                       return NULL;
-               }
+                       ereport(ERROR,
+                                       errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+                                       errmsg_internal("invalid opclass definition"),
+                                       errdetail_internal("The operator class is missing support function %d for column %d.",
+                                                                          procnum, attno));
        }
 
        return &opaque->extra_procinfos[basenum];

diff --git a/src/backend/access/brin/brin_inclusion.c b/src/backend/access/brin/brin_inclusion.c
index 4b02d374f2357d3b1468f1a313abd126a20d8a9e..c71e9cc5a6024c49ffb3c37fe2bbc40861c8f7b3 100644 (file)
--- a/src/backend/access/brin/brin_inclusion.c
+++ b/src/backend/access/brin/brin_inclusion.c
@@ -82,7 +82,7 @@ typedef struct InclusionOpaque
 } InclusionOpaque;
 
 static FmgrInfo *inclusion_get_procinfo(BrinDesc *bdesc, uint16 attno,
-                                                                               uint16 procnum);
+                                                                               uint16 procnum, bool missing_ok);
 static FmgrInfo *inclusion_get_strategy_procinfo(BrinDesc *bdesc, uint16 attno,
                                                                                                 Oid subtype, uint16 strategynum);
 
@@ -179,7 +179,7 @@ brin_inclusion_add_value(PG_FUNCTION_ARGS)
         * new value for emptiness; if it returns true, we need to set the
         * "contains empty" flag in the element (unless already set).
         */
-       finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_EMPTY);
+       finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_EMPTY, true);
        if (finfo != NULL && DatumGetBool(FunctionCall1Coll(finfo, colloid, newval)))
        {
                if (!DatumGetBool(column->bv_values[INCLUSION_CONTAINS_EMPTY]))
@@ -195,7 +195,7 @@ brin_inclusion_add_value(PG_FUNCTION_ARGS)
                PG_RETURN_BOOL(true);
 
        /* Check if the new value is already contained. */
-       finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_CONTAINS);
+       finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_CONTAINS, true);
        if (finfo != NULL &&
                DatumGetBool(FunctionCall2Coll(finfo, colloid,
                                                                           column->bv_values[INCLUSION_UNION],
@@ -210,7 +210,7 @@ brin_inclusion_add_value(PG_FUNCTION_ARGS)
         * it's not going to be used any longer.  However, the BRIN framework
         * doesn't allow for the value not being present.  Improve someday.
         */
-       finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_MERGEABLE);
+       finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_MERGEABLE, true);
        if (finfo != NULL &&
                !DatumGetBool(FunctionCall2Coll(finfo, colloid,
                                                                                column->bv_values[INCLUSION_UNION],
@@ -221,8 +221,7 @@ brin_inclusion_add_value(PG_FUNCTION_ARGS)
        }
 
        /* Finally, merge the new value to the existing union. */
-       finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_MERGE);
-       Assert(finfo != NULL);
+       finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_MERGE, false);
        result = FunctionCall2Coll(finfo, colloid,
                                                           column->bv_values[INCLUSION_UNION], newval);
        if (!attr->attbyval &&
@@ -506,7 +505,7 @@ brin_inclusion_union(PG_FUNCTION_ARGS)
        }
 
        /* Check if A and B are mergeable; if not, mark A unmergeable. */
-       finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_MERGEABLE);
+       finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_MERGEABLE, true);
        if (finfo != NULL &&
                !DatumGetBool(FunctionCall2Coll(finfo, colloid,
                                                                                col_a->bv_values[INCLUSION_UNION],
@@ -517,8 +516,7 @@ brin_inclusion_union(PG_FUNCTION_ARGS)
        }
 
        /* Finally, merge B to A. */
-       finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_MERGE);
-       Assert(finfo != NULL);
+       finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_MERGE, false);
        result = FunctionCall2Coll(finfo, colloid,
                                                           col_a->bv_values[INCLUSION_UNION],
                                                           col_b->bv_values[INCLUSION_UNION]);
@@ -539,10 +537,12 @@ brin_inclusion_union(PG_FUNCTION_ARGS)
  * Cache and return inclusion opclass support procedure
  *
  * Return the procedure corresponding to the given function support number
- * or null if it is not exists.
+ * or null if it is not exists.  If missing_ok is true and the procedure
+ * isn't set up for this opclass, return NULL instead of raising an error.
  */
 static FmgrInfo *
-inclusion_get_procinfo(BrinDesc *bdesc, uint16 attno, uint16 procnum)
+inclusion_get_procinfo(BrinDesc *bdesc, uint16 attno, uint16 procnum,
+                                          bool missing_ok)
 {
        InclusionOpaque *opaque;
        uint16          basenum = procnum - PROCNUM_BASE;
@@ -564,13 +564,18 @@ inclusion_get_procinfo(BrinDesc *bdesc, uint16 attno, uint16 procnum)
        {
                if (RegProcedureIsValid(index_getprocid(bdesc->bd_index, attno,
                                                                                                procnum)))
-               {
                        fmgr_info_copy(&opaque->extra_procinfos[basenum],
                                                   index_getprocinfo(bdesc->bd_index, attno, procnum),
                                                   bdesc->bd_context);
-               }
                else
                {
+                       if (!missing_ok)
+                               ereport(ERROR,
+                                               errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+                                               errmsg_internal("invalid opclass definition"),
+                                               errdetail_internal("The operator class is missing support function %d for column %d.",
+                                                                                  procnum, attno));
+
                        opaque->extra_proc_missing[basenum] = true;
                        return NULL;
                }

diff --git a/src/backend/access/brin/brin_minmax_multi.c b/src/backend/access/brin/brin_minmax_multi.c
index 9e29a08b121ee3ce41f63611261ed931865f454c..64f45aef3829f1002fde6f441409999aad6c1457 100644 (file)
--- a/src/backend/access/brin/brin_minmax_multi.c
+++ b/src/backend/access/brin/brin_minmax_multi.c
@@ -112,7 +112,6 @@
 typedef struct MinmaxMultiOpaque
 {
        FmgrInfo        extra_procinfos[MINMAX_MAX_PROCNUMS];
-       bool            extra_proc_missing[MINMAX_MAX_PROCNUMS];
        Oid                     cached_subtype;
        FmgrInfo        strategy_procinfos[BTMaxStrategyNumber];
 } MinmaxMultiOpaque;
@@ -2864,27 +2863,19 @@ minmax_multi_get_procinfo(BrinDesc *bdesc, uint16 attno, uint16 procnum)
         */
        opaque = (MinmaxMultiOpaque *) bdesc->bd_info[attno - 1]->oi_opaque;
 
-       /*
-        * If we already searched for this proc and didn't find it, don't bother
-        * searching again.
-        */
-       if (opaque->extra_proc_missing[basenum])
-               return NULL;
-
        if (opaque->extra_procinfos[basenum].fn_oid == InvalidOid)
        {
                if (RegProcedureIsValid(index_getprocid(bdesc->bd_index, attno,
                                                                                                procnum)))
-               {
                        fmgr_info_copy(&opaque->extra_procinfos[basenum],
                                                   index_getprocinfo(bdesc->bd_index, attno, procnum),
                                                   bdesc->bd_context);
-               }
                else
-               {
-                       opaque->extra_proc_missing[basenum] = true;
-                       return NULL;
-               }
+                       ereport(ERROR,
+                                       errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+                                       errmsg_internal("invalid opclass definition"),
+                                       errdetail_internal("The operator class is missing support function %d for column %d.",
+                                                                          procnum, attno));
        }
 
        return &opaque->extra_procinfos[basenum];