Apply selectivity correction to firstlast pairs

The compressed chunk size estimator already corrects Postgres's
column-independence assumption for orderby minmax pushdowns: per-batch
min and max are tightly correlated, so the estimator collapses the
predicate onto a single column to get accurate selectivity. After the
orderby default flipped to firstlast, predicates on the leading
NOT NULL orderby reference first/last metadata instead, and that
correction never fired for them.

Extend the same correction to firstlast pairs, direction-aware so
ASC and DESC both collapse onto a single metadata column.

Author: antekresic

tsl/src/compression/compression_scankey.c

@@ -224,13 +224,11 @@ build_heap_scankeys(Oid hypertable_relid, Relation in_rel, Relation out_rel,
 					continue;
 				}
 
+				/* Always use minmax metadata here. */
 				int16 index = ts_array_position(settings->fd.orderby, attname);
-				char *lower_name;
-				char *upper_name;
-				orderby_sparse_metadata_names(settings, index, &lower_name, &upper_name);
 
 				if (create_segment_filter_scankey(in_rel,
-												  lower_name,
+												  column_segment_min_name(index),
 												  BTLessEqualStrategyNumber,
 												  InvalidOid,
 												  InvalidOid,
@@ -246,7 +244,7 @@ build_heap_scankeys(Oid hypertable_relid, Relation in_rel, Relation out_rel,
 				}
 
 				if (create_segment_filter_scankey(in_rel,
-												  upper_name,
+												  column_segment_max_name(index),
 												  BTGreaterEqualStrategyNumber,
 												  InvalidOid,
 												  InvalidOid,

tsl/src/compression/create.c

@@ -659,7 +659,7 @@ build_columndefs(CompressionSettings *settings, Oid src_reloid)
 			/*
 			 * Every orderby column gets minmax metadata so range predicates
 			 * on any orderby position can be pushed down as a filter on the
-			 * compressed scan, regardless of NULL handling or direction.
+			 * compressed scan.
 			 */
 			ColumnDef *def = makeColumnDef(column_segment_min_name(index),
 										   attr->atttypid,
@@ -676,7 +676,7 @@ build_columndefs(CompressionSettings *settings, Oid src_reloid)
 
 			/*
 			 * When firstlast is configured for the orderby column, also add
-			 * first/last metadata. These columns lead the compressed-chunk
+			 * first/last metadata. These columns are a part of the compressed chunk
 			 * btree and let predicates on the leading orderby become index
 			 * conditions when the column is NOT NULL.
 			 */

tsl/src/nodes/columnar_scan/columnar_scan.c

@@ -399,22 +399,25 @@ copy_columnar_scan_path(ColumnarScanPath *src)
 }
 
 /*
- * Maps the attno of the min metadata column in the compressed chunk to the
- * attno of the corresponding max metadata column. Zero if none or not applicable.
+ * Maps the attno of a "lower" orderby metadata column on the compressed
+ * chunk to the attno of the corresponding "upper" column, and vice versa.
+ * Covers both minmax (lower=min, upper=max) and firstlast pairs.
+ * Zero entries mean none or not applicable.
  */
 typedef struct SelectivityEstimationContext
 {
-	AttrNumber *min_to_max;
-	AttrNumber *max_to_min;
+	AttrNumber *lower_to_upper;
+	AttrNumber *upper_to_lower;
 
 	List *vars;
 } SelectivityEstimationContext;
 
 /*
- * Collect the Vars referencing the "min" metadata columns into the context->vars.
+ * Collect the Vars referencing the orderby "lower" metadata columns
+ * (min for minmax pairs, first/last for firstlast pairs) into context->vars.
  */
 static bool
-min_metadata_vars_collector(Node *orig_node, SelectivityEstimationContext *context)
+lower_metadata_vars_collector(Node *orig_node, SelectivityEstimationContext *context)
 {
 	if (orig_node == NULL)
 	{
@@ -430,7 +433,7 @@ min_metadata_vars_collector(Node *orig_node, SelectivityEstimationContext *conte
 		/*
 		 * Recurse.
 		 */
-		return expression_tree_walker(orig_node, min_metadata_vars_collector, context);
+		return expression_tree_walker(orig_node, lower_metadata_vars_collector, context);
 	}
 
 	Var *orig_var = castNode(Var, orig_node);
@@ -442,7 +445,7 @@ min_metadata_vars_collector(Node *orig_node, SelectivityEstimationContext *conte
 		return false;
 	}
 
-	AttrNumber replaced_attno = context->min_to_max[orig_var->varattno];
+	AttrNumber replaced_attno = context->lower_to_upper[orig_var->varattno];
 	if (replaced_attno == InvalidAttrNumber)
 	{
 		/*
@@ -483,14 +486,14 @@ set_compressed_baserel_size_estimates(PlannerInfo *root, RelOptInfo *rel,
 	 * selectivity estimator must see the entire clause list to detect the range
 	 * conditions.
 	 *
-	 * First, build the correspondence of min metadata attno -> max metadata
-	 * attno for all minmax metadata.
+	 * First, build the correspondence of lower metadata attno -> upper
+	 * metadata attno for all orderby metadata pairs (minmax and firstlast).
 	 */
 	const int storage_elements = 2 * (compression_info->compressed_rel->max_attr + 1);
 	AttrNumber *storage = palloc0(storage_elements * sizeof(*storage));
 	SelectivityEstimationContext context = {
-		.min_to_max = &storage[0],
-		.max_to_min = &storage[compression_info->compressed_rel->max_attr],
+		.lower_to_upper = &storage[0],
+		.upper_to_lower = &storage[compression_info->compressed_rel->max_attr],
 	};
 
 	for (int uncompressed_attno = 1; uncompressed_attno <= compression_info->chunk_rel->max_attr;
@@ -534,42 +537,74 @@ set_compressed_baserel_size_estimates(PlannerInfo *root, RelOptInfo *rel,
 											 compression_info->compressed_rte->relid,
 											 "max");
 
-		if (min_attno == InvalidAttrNumber || max_attno == InvalidAttrNumber)
+		if (min_attno != InvalidAttrNumber && max_attno != InvalidAttrNumber)
 		{
-			continue;
+			Assert(&context.lower_to_upper[min_attno] < &storage[storage_elements]);
+			Assert(&context.upper_to_lower[max_attno] < &storage[storage_elements]);
+
+			context.lower_to_upper[min_attno] = max_attno;
+			context.upper_to_lower[max_attno] = min_attno;
 		}
 
-		Assert(&context.min_to_max[min_attno] < &storage[storage_elements]);
-		Assert(&context.max_to_min[max_attno] < &storage[storage_elements]);
+		/*
+		 * Same correlation hint for the firstlast pair when present. Under ASC
+		 * the lower bound is first and the upper is last; under DESC, the
+		 * roles flip. Mapping is direction-aware so that the Var swap below
+		 * collapses the predicate onto a single column regardless of shape.
+		 */
+		AttrNumber first_attno =
+			compressed_column_metadata_attno(compression_info->settings,
+											 compression_info->chunk_rte->relid,
+											 uncompressed_attno,
+											 compression_info->compressed_rte->relid,
+											 "first");
+		AttrNumber last_attno =
+			compressed_column_metadata_attno(compression_info->settings,
+											 compression_info->chunk_rte->relid,
+											 uncompressed_attno,
+											 compression_info->compressed_rte->relid,
+											 "last");
+
+		if (first_attno != InvalidAttrNumber && last_attno != InvalidAttrNumber)
+		{
+			bool desc =
+				ts_array_get_element_bool(compression_info->settings->fd.orderby_desc, orderby_pos);
+			AttrNumber lower_attno = desc ? last_attno : first_attno;
+			AttrNumber upper_attno = desc ? first_attno : last_attno;
+
+			Assert(&context.lower_to_upper[lower_attno] < &storage[storage_elements]);
+			Assert(&context.upper_to_lower[upper_attno] < &storage[storage_elements]);
 
-		context.min_to_max[min_attno] = max_attno;
-		context.max_to_min[max_attno] = min_attno;
+			context.lower_to_upper[lower_attno] = upper_attno;
+			context.upper_to_lower[upper_attno] = lower_attno;
+		}
 	}
 
 	/*
-	 * Then, replace all conditions on min metadata column with conditions on
-	 * max metadata column.
+	 * Then, collect all Var references to lower-bound metadata columns in the
+	 * restrict clauses so we can rewrite them to their upper-bound counterpart.
 	 */
 	ListCell *lc;
 	foreach (lc, rel->baserestrictinfo)
 	{
 		RestrictInfo *orig_restrictinfo = castNode(RestrictInfo, lfirst(lc));
 		Node *orig_clause = (Node *) orig_restrictinfo->clause;
-		expression_tree_walker(orig_clause, min_metadata_vars_collector, &context);
+		expression_tree_walker(orig_clause, lower_metadata_vars_collector, &context);
 	}
 
 	/*
-	 * Temporarily replace "min" with "max" in-place to save on memory allocations.
+	 * Temporarily replace lower-bound Vars with their upper-bound counterpart
+	 * in-place to save on memory allocations.
 	 */
 	foreach (lc, context.vars)
 	{
 		Var *var = castNode(Var, lfirst(lc));
 
 		Assert(var->varattno != InvalidAttrNumber);
-		Assert(context.min_to_max[var->varattno] != InvalidAttrNumber);
-		Assert(context.max_to_min[context.min_to_max[var->varattno]] == var->varattno);
+		Assert(context.lower_to_upper[var->varattno] != InvalidAttrNumber);
+		Assert(context.upper_to_lower[context.lower_to_upper[var->varattno]] == var->varattno);
 
-		var->varattno = context.min_to_max[var->varattno];
+		var->varattno = context.lower_to_upper[var->varattno];
 	}
 
 	/*
@@ -583,7 +618,7 @@ set_compressed_baserel_size_estimates(PlannerInfo *root, RelOptInfo *rel,
 	foreach (lc, context.vars)
 	{
 		Var *var = castNode(Var, lfirst(lc));
-		var->varattno = context.max_to_min[var->varattno];
+		var->varattno = context.upper_to_lower[var->varattno];
 	}
 
 	pfree(storage);

tsl/src/nodes/columnar_scan/qual_pushdown.c

@@ -183,13 +183,14 @@ make_segment_meta_opexpr(QualPushdownContext *context, Oid opno, AttrNumber meta
  * Returns InvalidAttrNumber via the out params when the expression is not a
  * sound pushdown target.
  *
- * For the leading orderby column under a firstlast-shaped chunk we prefer
- * the first/last metadata when the column is NOT NULL: those columns lead
- * the compressed-chunk btree, so the pushed-down predicate can become an
- * index condition. Every other case (nullable leading orderby, secondary
- * orderbys, non-orderby columns with an explicit minmax sparse index, and
- * any orderby on a legacy minmax-shaped chunk) falls back to minmax, which
- * is always available for orderby columns and direction-blind.
+ * For the leading orderby column under a firstlast-shaped compressed chunk
+ * index, we prefer the first/last metadata when the column is NOT NULL:
+ * those columns are part of the compressed chunk btree index, so the
+ * pushed-down predicate can become an index condition. Every other case
+ * (nullable leading orderby, secondary orderbys, non-orderby columns
+ * with an explicit minmax sparse index, and any orderby on a legacy
+ * minmax-shaped compressed chunk indexl) falls back to minmax, which
+ * is always available for orderby columns.
  */
 static void
 expr_fetch_orderby_range_metadata(QualPushdownContext *context, Expr *expr, AttrNumber *lower_attno,