Look up compressed column metadata only at planning time

Now we look them up again at execution time, which adds up for tables
with a large number of chunks.

This gives about 15% speedup (100 mcs) on a small query on a table from
tests with 50 chunks:
`select id, ts, value from metric_compressed order by id, ts limit 100;`

tsl/src/nodes/decompress_chunk/decompress_chunk.h

@@ -49,6 +49,17 @@ typedef struct DecompressChunkPath
 	 * uncompressed chunk, but are still used for decompression.
 	 */
 	List *decompression_map;
+
+	/*
+	 * This Int list is parallel to the compressed scan targetlist, just like
+	 * the above one. The value is true if a given targetlist entry is a
+	 * segmentby column, false otherwise. Has the same length as the above list.
+	 * We have to use the parallel lists and not a list of structs, because the
+	 * Plans have to be copyable by the Postgres _copy functions, and we can't
+	 * do that for a custom struct.
+	 */
+	List *is_segmentby_column;
+
 	List *compressed_pathkeys;
 	bool needs_sequence_num;
 	bool reverse;

tsl/src/nodes/decompress_chunk/exec.c

@@ -71,6 +71,7 @@ typedef struct DecompressChunkState
 {
 	CustomScanState csstate;
 	List *decompression_map;
+	List *is_segmentby_column;
 	int num_columns;
 	DecompressChunkColumnState *columns;
 
@@ -112,6 +113,7 @@ decompress_chunk_state_create(CustomScan *cscan)
 	state->chunk_relid = lsecond_int(settings);
 	state->reverse = lthird_int(settings);
 	state->decompression_map = lsecond(cscan->custom_private);
+	state->is_segmentby_column = lthird(cscan->custom_private);
 
 	return (Node *) state;
 }
@@ -128,7 +130,6 @@ initialize_column_state(DecompressChunkState *state)
 {
 	ScanState *ss = (ScanState *) state;
 	TupleDesc desc = ss->ss_ScanTupleSlot->tts_tupleDescriptor;
-	ListCell *lc;
 
 	if (list_length(state->decompression_map) == 0)
 	{
@@ -140,11 +141,14 @@ initialize_column_state(DecompressChunkState *state)
 
 	AttrNumber next_compressed_scan_attno = 0;
 	state->num_columns = 0;
-	foreach (lc, state->decompression_map)
+	ListCell *dest_cell;
+	ListCell *is_segmentby_cell;
+	Assert(list_length(state->decompression_map) == list_length(state->is_segmentby_column));
+	forboth (dest_cell, state->decompression_map, is_segmentby_cell, state->is_segmentby_column)
 	{
 		next_compressed_scan_attno++;
 
-		AttrNumber output_attno = lfirst_int(lc);
+		AttrNumber output_attno = lfirst_int(dest_cell);
 		if (output_attno == 0)
 		{
 			/* We are asked not to decompress this column, skip it. */
@@ -162,13 +166,10 @@ initialize_column_state(DecompressChunkState *state)
 			/* normal column that is also present in uncompressed chunk */
 			Form_pg_attribute attribute =
 				TupleDescAttr(desc, AttrNumberGetAttrOffset(output_attno));
-			FormData_hypertable_compression *ht_info =
-				get_column_compressioninfo(state->hypertable_compression_info,
-										   NameStr(attribute->attname));
 
 			column->typid = attribute->atttypid;
 
-			if (ht_info->segmentby_column_index > 0)
+			if (lfirst_int(is_segmentby_cell))
 				column->type = SEGMENTBY_COLUMN;
 			else
 				column->type = COMPRESSED_COLUMN;
@@ -290,8 +291,6 @@ decompress_chunk_begin(CustomScanState *node, EState *estate, int eflags)
 		}
 	}
 
-	state->hypertable_compression_info = ts_hypertable_compression_get(state->hypertable_id);
-
 	initialize_column_state(state);
 
 	node->custom_ps = lappend(node->custom_ps, ExecInitNode(compressed_scan, estate, eflags));

tsl/src/nodes/decompress_chunk/planner.c

@@ -227,6 +227,9 @@ build_decompression_map(DecompressChunkPath *path, List *scan_tlist, Bitmapset *
 
 		path->decompression_map =
 			lappend_int(path->decompression_map, destination_attno_in_uncompressed_chunk);
+		path->is_segmentby_column =
+			lappend_int(path->is_segmentby_column,
+						compression_info && compression_info->segmentby_column_index != 0);
 	}
 
 	/*
@@ -472,7 +475,8 @@ decompress_chunk_plan_create(PlannerInfo *root, RelOptInfo *rel, CustomPath *pat
 	settings = list_make3_int(dcpath->info->hypertable_id,
 							  dcpath->info->chunk_rte->relid,
 							  dcpath->reverse);
-	decompress_plan->custom_private = list_make2(settings, dcpath->decompression_map);
+	decompress_plan->custom_private =
+		list_make3(settings, dcpath->decompression_map, dcpath->is_segmentby_column);
 
 	return &decompress_plan->scan.plan;
 }