SkipScan over compressed chunks

Author: natalya-aksman

.unreleased/pr_7983

@@ -0,0 +1 @@
+Implements: #7983 Support for SkipScan over compressed data

src/guc.c

@@ -168,6 +168,7 @@ TSDLLEXPORT bool ts_guc_enable_skip_scan = true;
 #if PG16_GE
 TSDLLEXPORT bool ts_guc_enable_skip_scan_for_distinct_aggregates = true;
 #endif
+TSDLLEXPORT bool ts_guc_enable_compressed_skip_scan = true;
 static char *ts_guc_default_segmentby_fn = NULL;
 static char *ts_guc_default_orderby_fn = NULL;
 TSDLLEXPORT bool ts_guc_enable_job_execution_logging = false;
@@ -668,6 +669,18 @@ _guc_init(void)
 							 NULL,
 							 NULL);
 #endif
+
+	DefineCustomBoolVariable(MAKE_EXTOPTION("enable_compressed_skipscan"),
+							 "Enable SkipScan for compressed chunks",
+							 "Enable SkipScan for distinct inputs over compressed chunks",
+							 &ts_guc_enable_compressed_skip_scan,
+							 true,
+							 PGC_USERSET,
+							 0,
+							 NULL,
+							 NULL,
+							 NULL);
+
 	DefineCustomBoolVariable(MAKE_EXTOPTION("enable_compression_wal_markers"),
 							 "Enable WAL markers for compression ops",
 							 "Enable the generation of markers in the WAL stream which mark the "

src/guc.h

@@ -76,6 +76,7 @@ extern TSDLLEXPORT int ts_guc_compression_batch_size_limit;
 #if PG16_GE
 extern TSDLLEXPORT bool ts_guc_enable_skip_scan_for_distinct_aggregates;
 #endif
+extern TSDLLEXPORT bool ts_guc_enable_compressed_skip_scan;
 
 /* Only settable in debug mode for testing */
 extern TSDLLEXPORT bool ts_guc_enable_null_compression;

tsl/src/nodes/decompress_chunk/compressed_batch.c

@@ -1244,6 +1244,10 @@ compressed_batch_advance(DecompressContext *dcontext, DecompressBatchState *batc
 		}
 
 		/* The tuple passed the qual. */
+		if (dcontext->unique_segmentby)
+		{
+			batch_state->next_batch_row = batch_state->total_batch_rows - 1;
+		}
 		batch_state->next_batch_row++;
 		return;
 	}

tsl/src/nodes/decompress_chunk/decompress_chunk.h

@@ -61,5 +61,6 @@ void ts_decompress_chunk_generate_paths(PlannerInfo *root, RelOptInfo *rel, cons
 										const Chunk *chunk);
 
 extern bool ts_is_decompress_chunk_path(Path *path);
+extern bool ts_is_decompress_chunk_plan(Plan *plan);
 
 DecompressChunkPath *copy_decompress_chunk_path(DecompressChunkPath *src);

tsl/src/nodes/decompress_chunk/decompress_context.h

@@ -75,6 +75,7 @@ typedef struct DecompressContext
 	bool reverse;
 	bool batch_sorted_merge; /* Batch sorted merge optimization enabled. */
 	bool enable_bulk_decompression;
+	bool unique_segmentby;
 
 	/*
 	 * Scratch space for bulk decompression which might need a lot of temporary

tsl/src/nodes/decompress_chunk/exec.c

@@ -81,6 +81,7 @@ decompress_chunk_state_create(CustomScan *cscan)
 		list_nth_int(settings, DCS_BatchSortedMerge);
 	chunk_state->decompress_context.enable_bulk_decompression =
 		list_nth_int(settings, DCS_EnableBulkDecompression);
+	chunk_state->decompress_context.unique_segmentby = list_nth_int(settings, DCS_UniqueSegmentBy);
 	chunk_state->has_row_marks = list_nth_int(settings, DCS_HasRowMarks);
 
 	Assert(IsA(cscan->custom_exprs, List));

tsl/src/nodes/decompress_chunk/planner.c

@@ -44,6 +44,14 @@ static CustomScanMethods decompress_chunk_plan_methods = {
 	.CreateCustomScanState = decompress_chunk_state_create,
 };
 
+/* Check if the provided plan is a DecompressChunkPlan */
+bool
+ts_is_decompress_chunk_plan(Plan *plan)
+{
+	return IsA(plan, CustomScan) &&
+		   castNode(CustomScan, plan)->methods == &decompress_chunk_plan_methods;
+}
+
 void
 _decompress_chunk_init(void)
 {
@@ -1344,6 +1352,7 @@ decompress_chunk_plan_create(PlannerInfo *root, RelOptInfo *rel, CustomPath *pat
 	lfirst_int(list_nth_cell(settings, DCS_Reverse)) = dcpath->reverse;
 	lfirst_int(list_nth_cell(settings, DCS_BatchSortedMerge)) = dcpath->batch_sorted_merge;
 	lfirst_int(list_nth_cell(settings, DCS_EnableBulkDecompression)) = enable_bulk_decompression;
+	lfirst_int(list_nth_cell(settings, DCS_UniqueSegmentBy)) = false;
 	lfirst_int(list_nth_cell(settings, DCS_HasRowMarks)) = root->parse->rowMarks != NIL;
 
 	/*

tsl/src/nodes/decompress_chunk/planner.h

@@ -15,6 +15,7 @@ typedef enum
 	DCS_BatchSortedMerge = 3,
 	DCS_EnableBulkDecompression = 4,
 	DCS_HasRowMarks = 5,
+	DCS_UniqueSegmentBy = 6,
 	DCS_Count
 } DecompressChunkSettingsIndex;
 

tsl/src/nodes/skip_scan/exec.c

@@ -47,6 +47,7 @@
 #include <utils/datum.h>
 
 #include "guc.h"
+#include "nodes/decompress_chunk/decompress_chunk.h"
 #include "nodes/skip_scan/skip_scan.h"
 
 typedef enum SkipScanStage
@@ -88,6 +89,12 @@ typedef struct SkipScanState
 	/* rescan required before getting next tuple */
 	bool needs_rescan;
 
+	/* child_plan node is the input for skip scan plan node.
+	 * if skipscan is directly over index, child_plan = idx_scan
+	 * if skip scan is over compressed index, idx_scan = compressed_index,
+	 * child_plan = decompressed input into skip scan
+	 */
+	Plan *child_plan;
 	void *idx_scan;
 } SkipScanState;
 
@@ -102,8 +109,19 @@ skip_scan_begin(CustomScanState *node, EState *estate, int eflags)
 	SkipScanState *state = (SkipScanState *) node;
 	state->ctx = AllocSetContextCreate(estate->es_query_cxt, "skipscan", ALLOCSET_DEFAULT_SIZES);
 
-	state->idx = (ScanState *) ExecInitNode(state->idx_scan, estate, eflags);
-	node->custom_ps = list_make1(state->idx);
+	node->custom_ps = list_make1((ScanState *) ExecInitNode(state->child_plan, estate, eflags));
+	ScanState *child_state = linitial(node->custom_ps);
+	if (state->child_plan == state->idx_scan)
+	{
+		state->idx = child_state;
+	}
+	else if (IsA(child_state, CustomScanState))
+	{
+		Assert(ts_is_decompress_chunk_plan(state->child_plan));
+		state->idx = linitial(castNode(CustomScanState, child_state)->custom_ps);
+	}
+	else
+		elog(ERROR, "unknown subscan type in SkipScan");
 
 	if (IsA(state->idx_scan, IndexScan))
 	{
@@ -242,6 +260,7 @@ skip_scan_exec(CustomScanState *node)
 {
 	SkipScanState *state = (SkipScanState *) node;
 	TupleTableSlot *result;
+	ScanState *child_state;
 
 	/*
 	 * We are not supporting projection here since no plan
@@ -267,7 +286,8 @@ skip_scan_exec(CustomScanState *node)
 				break;
 
 			case SS_NULLS_FIRST:
-				result = state->idx->ps.ExecProcNode(&state->idx->ps);
+				child_state = linitial(state->cscan_state.custom_ps);
+				result = child_state->ps.ExecProcNode(&child_state->ps);
 
 				/*
 				 * if we found a NULL value we return it, otherwise
@@ -281,7 +301,8 @@ skip_scan_exec(CustomScanState *node)
 
 			case SS_NOT_NULL:
 			case SS_VALUES:
-				result = state->idx->ps.ExecProcNode(&state->idx->ps);
+				child_state = linitial(state->cscan_state.custom_ps);
+				result = child_state->ps.ExecProcNode(&child_state->ps);
 
 				if (!TupIsNull(result))
 				{
@@ -314,7 +335,8 @@ skip_scan_exec(CustomScanState *node)
 				break;
 
 			case SS_NULLS_LAST:
-				result = state->idx->ps.ExecProcNode(&state->idx->ps);
+				child_state = linitial(state->cscan_state.custom_ps);
+				result = child_state->ps.ExecProcNode(&child_state->ps);
 				skip_scan_switch_stage(state, SS_END);
 				return result;
 				break;
@@ -330,7 +352,8 @@ static void
 skip_scan_end(CustomScanState *node)
 {
 	SkipScanState *state = (SkipScanState *) node;
-	ExecEndNode(&state->idx->ps);
+	ScanState *child_state = linitial(state->cscan_state.custom_ps);
+	ExecEndNode(&child_state->ps);
 }
 
 static void
@@ -352,7 +375,8 @@ skip_scan_rescan(CustomScanState *node)
 	state->prev_datum = 0;
 
 	state->needs_rescan = false;
-	ExecReScan(&state->idx->ps);
+	ScanState *child_state = linitial(state->cscan_state.custom_ps);
+	ExecReScan(&child_state->ps);
 	MemoryContextReset(state->ctx);
 }
 
@@ -369,7 +393,16 @@ tsl_skip_scan_state_create(CustomScan *cscan)
 {
 	SkipScanState *state = (SkipScanState *) newNode(sizeof(SkipScanState), T_CustomScanState);
 
-	state->idx_scan = linitial(cscan->custom_plans);
+	state->child_plan = linitial(cscan->custom_plans);
+	if (ts_is_decompress_chunk_plan(state->child_plan))
+	{
+		CustomScan *csplan = castNode(CustomScan, state->child_plan);
+		state->idx_scan = linitial(csplan->custom_plans);
+	}
+	else
+	{
+		state->idx_scan = state->child_plan;
+	}
 	state->stage = SS_BEGIN;
 
 	state->distinct_col_attnum = linitial_int(cscan->custom_private);