Change signature of ts_chunk_tuple_routing_create

Change signature of ts_chunk_tuple_routing_create in preparation
for ChunkDispatch removal. This patch extracts the COPY specific
changes.

Author: svenklemm

src/chunk_tuple_routing.c

@@ -10,12 +10,13 @@
 #include "debug_point.h"
 #include "hypercube.h"
 #include "nodes/chunk_dispatch/chunk_insert_state.h"
+#include "nodes/modify_hypertable.h"
 #include "subspace_store.h"
 
 static ChunkInsertState *chunk_insert_state_create(Oid chunk_relid, ChunkTupleRouting *ctr);
 
 ChunkTupleRouting *
-ts_chunk_tuple_routing_create(EState *estate, Relation rel)
+ts_chunk_tuple_routing_create(EState *estate, ResultRelInfo *rri)
 {
 	ChunkTupleRouting *ctr;
 
@@ -27,14 +28,16 @@ ts_chunk_tuple_routing_create(EState *estate, Relation rel)
 	 * single tuple into a partitioned table and this must be fast.
 	 */
 	ctr = (ChunkTupleRouting *) palloc0(sizeof(ChunkTupleRouting));
-	ctr->partition_root = rel;
+	ctr->hypertable_rri = rri;
+	ctr->partition_root = rri->ri_RelationDesc;
 	ctr->memcxt = CurrentMemoryContext;
 	ctr->estate = estate;
 	ctr->counters = palloc0(sizeof(SharedCounters));
 
-	ctr->hypertable = ts_hypertable_cache_get_cache_and_entry(RelationGetRelid(rel),
-															  CACHE_FLAG_NONE,
-															  &ctr->hypertable_cache);
+	ctr->hypertable =
+		ts_hypertable_cache_get_cache_and_entry(RelationGetRelid(rri->ri_RelationDesc),
+												CACHE_FLAG_NONE,
+												&ctr->hypertable_cache);
 	ctr->subspace = ts_subspace_store_init(ctr->hypertable->space,
 										   estate->es_query_cxt,
 										   ts_guc_max_open_chunks_per_insert);
@@ -248,3 +251,42 @@ chunk_insert_state_create(Oid chunk_relid, ChunkTupleRouting *ctr)
 
 	return state;
 }
+
+extern void
+ts_chunk_tuple_routing_decompress_for_insert(ChunkInsertState *cis, TupleTableSlot *slot,
+											 EState *estate, bool update_counter)
+{
+	if (!cis->chunk_compressed || (cis->cached_decompression_state &&
+								   !cis->cached_decompression_state->has_primary_or_unique_index))
+		return;
+
+	/*
+	 * If this is an INSERT into a compressed chunk with UNIQUE or
+	 * PRIMARY KEY constraints we need to make sure any batches that could
+	 * potentially lead to a conflict are in the decompressed chunk so
+	 * postgres can do proper constraint checking.
+	 */
+
+	ts_cm_functions->init_decompress_state_for_insert(cis, slot);
+	ts_cm_functions->decompress_batches_for_insert(cis, slot);
+
+	/* mark rows visible */
+	if (update_counter)
+		estate->es_output_cid = GetCurrentCommandId(true);
+
+	if (ts_guc_max_tuples_decompressed_per_dml > 0)
+	{
+		if (cis->counters->tuples_decompressed > ts_guc_max_tuples_decompressed_per_dml)
+		{
+			ereport(ERROR,
+					(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
+					 errmsg("tuple decompression limit exceeded by operation"),
+					 errdetail("current limit: %d, tuples decompressed: %lld",
+							   ts_guc_max_tuples_decompressed_per_dml,
+							   (long long int) cis->counters->tuples_decompressed),
+					 errhint("Consider increasing "
+							 "timescaledb.max_tuples_decompressed_per_dml_transaction or set "
+							 "to 0 (unlimited).")));
+		}
+	}
+}

src/chunk_tuple_routing.h

@@ -29,6 +29,9 @@ typedef struct ChunkTupleRouting
 	SharedCounters *counters; /* shared counters for the current statement */
 } ChunkTupleRouting;
 
-ChunkTupleRouting *ts_chunk_tuple_routing_create(EState *estate, Relation rel);
+ChunkTupleRouting *ts_chunk_tuple_routing_create(EState *estate, ResultRelInfo *rri);
 void ts_chunk_tuple_routing_destroy(ChunkTupleRouting *ctr);
 ChunkInsertState *ts_chunk_tuple_routing_find_chunk(ChunkTupleRouting *ctr, Point *point);
+extern void ts_chunk_tuple_routing_decompress_for_insert(ChunkInsertState *cis,
+														 TupleTableSlot *slot, EState *estate,
+														 bool update_counter);

src/copy.c

@@ -161,7 +161,6 @@ copy_chunk_state_create(Hypertable *ht, Relation rel, CopyFromFunc from_func, Co
 	ccstate = palloc(sizeof(CopyChunkState));
 	ccstate->rel = rel;
 	ccstate->estate = estate;
-	ccstate->ctr = ts_chunk_tuple_routing_create(estate, rel);
 	ccstate->cstate = cstate;
 	ccstate->scandesc = scandesc;
 	ccstate->next_copy_from = from_func;
@@ -890,7 +889,7 @@ copyfrom(CopyChunkState *ccstate, ParseState *pstate, Hypertable *ht, MemoryCont
 
 	ExecOpenIndices(resultRelInfo, false);
 
-	ccstate->ctr->hypertable_rri = resultRelInfo;
+	ccstate->ctr = ts_chunk_tuple_routing_create(estate, resultRelInfo);
 
 	singleslot = table_slot_create(resultRelInfo->ri_RelationDesc, &estate->es_tupleTable);
 
@@ -1036,7 +1035,7 @@ copyfrom(CopyChunkState *ccstate, ParseState *pstate, Hypertable *ht, MemoryCont
 
 		Assert(cis != NULL);
 
-		ts_chunk_dispatch_decompress_batches_for_insert(cis, myslot, ccstate->ctr->estate, false);
+		ts_chunk_tuple_routing_decompress_for_insert(cis, myslot, ccstate->ctr->estate, false);
 
 		/* Triggers and stuff need to be invoked in query context. */
 		MemoryContextSwitchTo(oldcontext);

src/nodes/chunk_dispatch/chunk_dispatch.c

@@ -20,6 +20,7 @@
 #include "compat/compat.h"
 #include "chunk_dispatch.h"
 #include "chunk_insert_state.h"
+#include "chunk_tuple_routing.h"
 #include "dimension.h"
 #include "errors.h"
 #include "guc.h"
@@ -173,45 +174,6 @@ ts_chunk_dispatch_get_chunk_insert_state(ChunkDispatch *dispatch, Point *point,
 	return cis;
 }
 
-extern void
-ts_chunk_dispatch_decompress_batches_for_insert(ChunkInsertState *cis, TupleTableSlot *slot,
-												EState *estate, bool update_counter)
-{
-	if (!cis->chunk_compressed || (cis->cached_decompression_state &&
-								   !cis->cached_decompression_state->has_primary_or_unique_index))
-		return;
-
-	/*
-	 * If this is an INSERT into a compressed chunk with UNIQUE or
-	 * PRIMARY KEY constraints we need to make sure any batches that could
-	 * potentially lead to a conflict are in the decompressed chunk so
-	 * postgres can do proper constraint checking.
-	 */
-
-	ts_cm_functions->init_decompress_state_for_insert(cis, slot);
-	ts_cm_functions->decompress_batches_for_insert(cis, slot);
-
-	/* mark rows visible */
-	if (update_counter)
-		estate->es_output_cid = GetCurrentCommandId(true);
-
-	if (ts_guc_max_tuples_decompressed_per_dml > 0)
-	{
-		if (cis->counters->tuples_decompressed > ts_guc_max_tuples_decompressed_per_dml)
-		{
-			ereport(ERROR,
-					(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
-					 errmsg("tuple decompression limit exceeded by operation"),
-					 errdetail("current limit: %d, tuples decompressed: %lld",
-							   ts_guc_max_tuples_decompressed_per_dml,
-							   (long long int) cis->counters->tuples_decompressed),
-					 errhint("Consider increasing "
-							 "timescaledb.max_tuples_decompressed_per_dml_transaction or set "
-							 "to 0 (unlimited).")));
-		}
-	}
-}
-
 static CustomScanMethods chunk_dispatch_plan_methods = {
 	.CustomName = "ChunkDispatch",
 	.CreateCustomScanState = chunk_dispatch_state_create,
@@ -433,10 +395,7 @@ chunk_dispatch_exec(CustomScanState *node)
 		bool update_counter =
 			ts_chunk_dispatch_get_on_conflict_action(dispatch) == ONCONFLICT_UPDATE;
 
-		ts_chunk_dispatch_decompress_batches_for_insert(cis,
-														slot,
-														dispatch->estate,
-														update_counter);
+		ts_chunk_tuple_routing_decompress_for_insert(cis, slot, dispatch->estate, update_counter);
 	}
 
 	MemoryContextSwitchTo(old);