Reduce decompression for INSERT with unique constraints

On INSERT into compressed chunks with unique constraints we can
check for conflict without decompressing when no ON CONFLICT clause
is present and we only have one unique constraint. With ON CONFLICT
clause with DO NOTHING we can just skip the INSERT if we detect conflict
and return early. Only for ON CONFLICT DO UPDATE/UPSERT do we need
to decompress when there is a constraint conflict.
Doing the optimization in the presence of multiple constraints is
also possible but not part of this patch.

.unreleased/pr_7108

@@ -0,0 +1 @@
+Implements: #7108 Reduce decompressions for INSERTs with UNIQUE constraints

src/nodes/chunk_dispatch/chunk_dispatch.c

@@ -173,7 +173,7 @@ ts_chunk_dispatch_decompress_batches_for_insert(ChunkDispatch *dispatch, ChunkIn
 		if (ts_cm_functions->decompress_batches_for_insert)
 		{
 			ts_cm_functions->decompress_batches_for_insert(cis, slot);
-			OnConflictAction onconflict_action = chunk_dispatch_get_on_conflict_action(dispatch);
+			OnConflictAction onconflict_action = ts_chunk_dispatch_get_on_conflict_action(dispatch);
 			/* mark rows visible */
 			if (onconflict_action == ONCONFLICT_UPDATE)
 				dispatch->estate->es_output_cid = GetCurrentCommandId(true);

src/nodes/chunk_dispatch/chunk_dispatch.h

@@ -76,6 +76,9 @@ typedef struct ChunkDispatchState
 	bool is_dropped_attr_exists;
 	int64 batches_decompressed;
 	int64 tuples_decompressed;
+
+	/* Should this INSERT be skipped due to ON CONFLICT DO NOTHING */
+	bool skip_current_tuple;
 } ChunkDispatchState;
 
 extern TSDLLEXPORT bool ts_is_chunk_dispatch_state(PlanState *state);

src/nodes/chunk_dispatch/chunk_insert_state.c

@@ -80,7 +80,7 @@ chunk_dispatch_get_returning_clauses(const ChunkDispatch *dispatch)
 }
 
 OnConflictAction
-chunk_dispatch_get_on_conflict_action(const ChunkDispatch *dispatch)
+ts_chunk_dispatch_get_on_conflict_action(const ChunkDispatch *dispatch)
 {
 	if (!dispatch->dispatch_state || !dispatch->dispatch_state->mtstate)
 		return ONCONFLICT_NONE;
@@ -265,7 +265,7 @@ setup_on_conflict_state(ChunkInsertState *state, const ChunkDispatch *dispatch,
 	ModifyTableState *mtstate = castNode(ModifyTableState, dispatch->dispatch_state->mtstate);
 	ModifyTable *mt = castNode(ModifyTable, mtstate->ps.plan);
 
-	Assert(chunk_dispatch_get_on_conflict_action(dispatch) == ONCONFLICT_UPDATE);
+	Assert(ts_chunk_dispatch_get_on_conflict_action(dispatch) == ONCONFLICT_UPDATE);
 
 	OnConflictSetState *onconfl = makeNode(OnConflictSetState);
 	memcpy(onconfl, hyper_rri->ri_onConflict, sizeof(OnConflictSetState));
@@ -433,7 +433,7 @@ adjust_projections(ChunkInsertState *cis, const ChunkDispatch *dispatch, Oid row
 	Relation hyper_rel = dispatch->hypertable_result_rel_info->ri_RelationDesc;
 	Relation chunk_rel = cis->rel;
 	TupleConversionMap *chunk_map = NULL;
-	OnConflictAction onconflict_action = chunk_dispatch_get_on_conflict_action(dispatch);
+	OnConflictAction onconflict_action = ts_chunk_dispatch_get_on_conflict_action(dispatch);
 
 	if (chunk_dispatch_has_returning(dispatch))
 	{
@@ -479,7 +479,7 @@ ts_chunk_insert_state_create(Oid chunk_relid, const ChunkDispatch *dispatch)
 	MemoryContext cis_context = AllocSetContextCreate(dispatch->estate->es_query_cxt,
 													  "chunk insert state memory context",
 													  ALLOCSET_DEFAULT_SIZES);
-	OnConflictAction onconflict_action = chunk_dispatch_get_on_conflict_action(dispatch);
+	OnConflictAction onconflict_action = ts_chunk_dispatch_get_on_conflict_action(dispatch);
 	ResultRelInfo *relinfo;
 	const Chunk *chunk;
 

src/nodes/chunk_dispatch/chunk_insert_state.h

@@ -65,5 +65,6 @@ extern ChunkInsertState *ts_chunk_insert_state_create(Oid chunk_relid,
 													  const ChunkDispatch *dispatch);
 extern void ts_chunk_insert_state_destroy(ChunkInsertState *state);
 
-OnConflictAction chunk_dispatch_get_on_conflict_action(const ChunkDispatch *dispatch);
+TSDLLEXPORT OnConflictAction
+ts_chunk_dispatch_get_on_conflict_action(const ChunkDispatch *dispatch);
 void ts_set_compression_status(ChunkInsertState *state, const Chunk *chunk);

src/nodes/hypertable_modify.c

@@ -687,6 +687,14 @@ ExecModifyTable(CustomScanState *cs_node, PlanState *pstate)
 
 		context.planSlot = ExecProcNode(subplanstate);
 
+		if (cds && cds->rri && operation == CMD_INSERT && cds->skip_current_tuple)
+		{
+			cds->skip_current_tuple = false;
+			if (node->ps.instrument)
+				node->ps.instrument->ntuples2++;
+			return NULL;
+		}
+
 		/* No more tuples to process? */
 		if (TupIsNull(context.planSlot))
 			break;