Add dimension, chunk entries for correlated constraint

Create dimension_slice and chunk_constraint entries for chunks which
have correlated constraints on them. The dimension slice entry will
have -inf/+inf as start/end range initially for a given correlated
constraint and the chunk_constraint entry will refer back to this slice
entry.

This start/end range will be refreshed later. One of the entry points
is during compression for now.

src/chunk.c

@@ -841,14 +841,16 @@ chunk_create_object(const Hypertable *ht, Hypercube *cube, const char *schema_na
 					const char *table_name, const char *prefix, int32 chunk_id)
 {
 	const Hyperspace *hs = ht->space;
+	const Hyperspace *ss = ht->correlated_space;
 	Chunk *chunk;
 	const char relkind = RELKIND_RELATION;
 
 	if (NULL == schema_name || schema_name[0] == '\0')
 		schema_name = NameStr(ht->fd.associated_schema_name);
 
 	/* Create a new chunk based on the hypercube */
-	chunk = ts_chunk_create_base(chunk_id, hs->num_dimensions, relkind);
+	chunk =
+		ts_chunk_create_base(chunk_id, hs->num_dimensions + (ss ? ss->num_dimensions : 0), relkind);
 
 	chunk->fd.hypertable_id = hs->hypertable_id;
 	chunk->cube = cube;
@@ -1091,6 +1093,9 @@ chunk_create_from_hypercube_after_lock(const Hypertable *ht, Hypercube *cube,
 													  prefix,
 													  get_next_chunk_id());
 
+	/* Insert any new correlated constraint slices into metadata */
+	ts_correlated_constraints_dimension_slice_insert(ht, chunk);
+
 	chunk_add_constraints(chunk);
 	chunk_insert_into_metadata_after_lock(chunk);
 	chunk_create_table_constraints(ht, chunk);

src/chunk_adaptive.c

@@ -266,8 +266,9 @@ table_has_minmax_index(Oid relid, Oid atttype, Name attname, AttrNumber attnum)
  *
  * Returns true iff min and max is found, otherwise false.
  */
-static bool
-chunk_get_minmax(Oid relid, Oid atttype, AttrNumber attnum, Datum minmax[2])
+bool
+ts_chunk_get_minmax(Oid relid, Oid atttype, AttrNumber attnum, const char *call_context,
+					Datum minmax[2])
 {
 	Relation rel = table_open(relid, AccessShareLock);
 	NameData attname;
@@ -279,11 +280,11 @@ chunk_get_minmax(Oid relid, Oid atttype, AttrNumber attnum, Datum minmax[2])
 	if (res == MINMAX_NO_INDEX)
 	{
 		ereport(WARNING,
-				(errmsg("no index on \"%s\" found for adaptive chunking on chunk \"%s\"",
+				(errmsg("no index on \"%s\" found for %s on chunk \"%s\"",
+						call_context,
 						NameStr(attname),
 						get_rel_name(relid)),
-				 errdetail("Adaptive chunking works best with an index on the dimension being "
-						   "adapted.")));
+				 errdetail("%s works best with an index on the dimension.", call_context)));
 
 		res = minmax_heapscan(rel, atttype, attnum, minmax);
 	}
@@ -469,7 +470,11 @@ ts_calculate_chunk_interval(PG_FUNCTION_ARGS)
 
 		slice_interval = slice->fd.range_end - slice->fd.range_start;
 
-		if (chunk_get_minmax(chunk->table_id, dim->fd.column_type, attno, minmax))
+		if (ts_chunk_get_minmax(chunk->table_id,
+								dim->fd.column_type,
+								attno,
+								"adaptive chunking",
+								minmax))
 		{
 			int64 min = ts_time_value_to_internal(minmax[0], dim->fd.column_type);
 			int64 max = ts_time_value_to_internal(minmax[1], dim->fd.column_type);

src/chunk_adaptive.h

@@ -24,6 +24,8 @@ typedef struct ChunkSizingInfo
 
 extern void ts_chunk_adaptive_sizing_info_validate(ChunkSizingInfo *info);
 extern void ts_chunk_sizing_func_validate(regproc func, ChunkSizingInfo *info);
+extern bool ts_chunk_get_minmax(Oid relid, Oid atttype, AttrNumber attnum, const char *call_context,
+								Datum minmax[2]);
 extern TSDLLEXPORT ChunkSizingInfo *ts_chunk_sizing_info_get_default_disabled(Oid table_relid);
 
 extern TSDLLEXPORT int64 ts_chunk_calculate_initial_chunk_target_size(void);

src/chunk_constraint.c

@@ -88,9 +88,9 @@ chunk_constraints_expand(ChunkConstraints *ccs, int16 new_capacity)
 }
 
 static void
-chunk_constraint_dimension_choose_name(Name dst, int32 dimension_slice_id)
+chunk_constraint_dimension_choose_name(Name dst, const char *prefix, int32 dimension_slice_id)
 {
-	snprintf(NameStr(*dst), NAMEDATALEN, "constraint_%d", dimension_slice_id);
+	snprintf(NameStr(*dst), NAMEDATALEN, "%sconstraint_%d", prefix, dimension_slice_id);
 }
 
 static void
@@ -115,20 +115,27 @@ chunk_constraint_choose_name(Name dst, const char *hypertable_constraint_name, i
 
 ChunkConstraint *
 ts_chunk_constraints_add(ChunkConstraints *ccs, int32 chunk_id, int32 dimension_slice_id,
-						 const char *constraint_name, const char *hypertable_constraint_name)
+						 const char *constraint_name, const char *hypertable_constraint_name,
+						 bool correlated)
 {
 	ChunkConstraint *cc;
 
 	chunk_constraints_expand(ccs, ccs->num_constraints + 1);
 	cc = &ccs->constraints[ccs->num_constraints++];
 	cc->fd.chunk_id = chunk_id;
 	cc->fd.dimension_slice_id = dimension_slice_id;
+	cc->correlated = correlated;
 
 	if (NULL == constraint_name)
 	{
-		if (is_dimension_constraint(cc))
+		if (is_dimension_constraint(cc) || correlated)
 		{
+			/*
+			 * for correlated constraints we choose a prefix of "_$CC_" to help
+			 * us identify it on re-reading from the catalog.
+			 */
 			chunk_constraint_dimension_choose_name(&cc->fd.constraint_name,
+												   correlated ? CC_DIM_PREFIX : "",
 												   cc->fd.dimension_slice_id);
 			namestrcpy(&cc->fd.hypertable_constraint_name, "");
 		}
@@ -163,7 +170,7 @@ chunk_constraint_fill_tuple_values(const ChunkConstraint *cc, Datum values[Natts
 	values[AttrNumberGetAttrOffset(Anum_chunk_constraint_hypertable_constraint_name)] =
 		NameGetDatum(&cc->fd.hypertable_constraint_name);
 
-	if (is_dimension_constraint(cc))
+	if (is_dimension_constraint(cc) || cc->correlated)
 		nulls[AttrNumberGetAttrOffset(Anum_chunk_constraint_hypertable_constraint_name)] = true;
 	else
 		nulls[AttrNumberGetAttrOffset(Anum_chunk_constraint_dimension_slice_id)] = true;
@@ -229,7 +236,7 @@ ts_chunk_constraints_add_from_tuple(ChunkConstraints *ccs, const TupleInfo *ti)
 	int32 dimension_slice_id;
 	Name constraint_name;
 	Name hypertable_constraint_name;
-	bool should_free;
+	bool should_free, correlated = false;
 	HeapTuple tuple = ts_scanner_fetch_heap_tuple(ti, false, &should_free);
 	MemoryContext oldcxt;
 
@@ -253,12 +260,17 @@ ts_chunk_constraints_add_from_tuple(ChunkConstraints *ccs, const TupleInfo *ti)
 		hypertable_constraint_name = DatumGetName(DirectFunctionCall1(namein, CStringGetDatum("")));
 	}
 
+	/* check if it's a correlated constraint */
+	if (strncmp(CC_DIM_PREFIX, NameStr(*constraint_name), strlen(CC_DIM_PREFIX)) == 0)
+		correlated = true;
+
 	constraints = ts_chunk_constraints_add(ccs,
 										   DatumGetInt32(values[AttrNumberGetAttrOffset(
 											   Anum_chunk_constraint_chunk_id)]),
 										   dimension_slice_id,
 										   NameStr(*constraint_name),
-										   NameStr(*hypertable_constraint_name));
+										   NameStr(*hypertable_constraint_name),
+										   correlated);
 
 	MemoryContextSwitchTo(oldcxt);
 
@@ -492,7 +504,7 @@ ts_chunk_constraints_create(const Hypertable *ht, const Chunk *chunk)
 			if (constr != NULL)
 				newconstrs = lappend(newconstrs, constr);
 		}
-		else
+		else if (!cc->correlated)
 		{
 			create_non_dimensional_constraint(cc,
 											  chunk->table_id,
@@ -781,7 +793,7 @@ ts_chunk_constraints_add_dimension_constraints(ChunkConstraints *ccs, int32 chun
 	int i;
 
 	for (i = 0; i < cube->num_slices; i++)
-		ts_chunk_constraints_add(ccs, chunk_id, cube->slices[i]->fd.id, NULL, NULL);
+		ts_chunk_constraints_add(ccs, chunk_id, cube->slices[i]->fd.id, NULL, NULL, false);
 
 	return cube->num_slices;
 }
@@ -802,7 +814,17 @@ chunk_constraint_add(HeapTuple constraint_tuple, void *arg)
 
 	if (chunk_constraint_need_on_chunk(cc->chunk_relkind, constraint))
 	{
-		ts_chunk_constraints_add(cc->ccs, cc->chunk_id, 0, NULL, NameStr(constraint->conname));
+		bool correlated = false;
+
+		/* check if it's a correlated constraint */
+		if (strncmp(CC_DIM_PREFIX, NameStr(constraint->conname), strlen(CC_DIM_PREFIX)) == 0)
+			correlated = true;
+		ts_chunk_constraints_add(cc->ccs,
+								 cc->chunk_id,
+								 0,
+								 NULL,
+								 NameStr(constraint->conname),
+								 correlated);
 		return CONSTR_PROCESSED;
 	}
 
@@ -835,7 +857,8 @@ chunk_constraint_add_check(HeapTuple constraint_tuple, void *arg)
 								 cc->chunk_id,
 								 0,
 								 NameStr(constraint->conname),
-								 NameStr(constraint->conname));
+								 NameStr(constraint->conname),
+								 false);
 		return CONSTR_PROCESSED;
 	}
 
@@ -874,7 +897,8 @@ ts_chunk_constraint_create_on_chunk(const Hypertable *ht, const Chunk *chunk, Oi
 													   chunk->fd.id,
 													   0,
 													   NULL,
-													   NameStr(con->conname));
+													   NameStr(con->conname),
+													   false);
 
 		ts_chunk_constraint_insert(cc);
 		create_non_dimensional_constraint(cc,

src/chunk_constraint.h

@@ -11,9 +11,11 @@
 #include "ts_catalog/catalog.h"
 #include "hypertable.h"
 
+#define CC_DIM_PREFIX "_$CC_" /* Correlated Constraint dimension */
 typedef struct ChunkConstraint
 {
 	FormData_chunk_constraint fd;
+	bool correlated;
 } ChunkConstraint;
 
 typedef struct ChunkConstraints
@@ -27,7 +29,8 @@ typedef struct ChunkConstraints
 
 #define chunk_constraints_get(cc, i) &((cc)->constraints[i])
 
-#define is_dimension_constraint(cc) ((cc)->fd.dimension_slice_id > 0)
+#define is_dimension_constraint(cc) ((cc)->fd.dimension_slice_id > 0 && !(cc)->correlated)
+#define is_correlated_constraint_dimension(cc) (cc)->correlated
 
 typedef struct Chunk Chunk;
 typedef struct DimensionSlice DimensionSlice;
@@ -48,7 +51,8 @@ extern int ts_chunk_constraint_scan_by_dimension_slice_id(int32 dimension_slice_
 extern ChunkConstraint *ts_chunk_constraints_add(ChunkConstraints *ccs, int32 chunk_id,
 												 int32 dimension_slice_id,
 												 const char *constraint_name,
-												 const char *hypertable_constraint_name);
+												 const char *hypertable_constraint_name,
+												 bool correlated);
 extern int ts_chunk_constraints_add_dimension_constraints(ChunkConstraints *ccs, int32 chunk_id,
 														  const Hypercube *cube);
 extern TSDLLEXPORT int ts_chunk_constraints_add_inheritable_constraints(ChunkConstraints *ccs,

src/dimension.c

@@ -1636,12 +1636,18 @@ ts_dimension_add_internal(FunctionCallInfo fcinfo, DimensionInfo *info, bool is_
 
 	info->ht = ts_hypertable_cache_get_cache_and_entry(info->table_relid, CACHE_FLAG_NONE, &hcache);
 
+	if (info->correlated)
+	{
+		info->interval_type = INT8OID;
+		info->interval_datum = DatumGetInt32(1);
+	}
+
 	if (info->num_slices_is_set && OidIsValid(info->interval_type))
 		ereport(ERROR,
 				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 				 errmsg("cannot specify both the number of partitions and an interval")));
 
-	if (!info->num_slices_is_set && !OidIsValid(info->interval_type) && !info->correlated)
+	if (!info->num_slices_is_set && !OidIsValid(info->interval_type))
 		ereport(ERROR,
 				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 				 errmsg("cannot omit both the number of partitions and the interval")));
@@ -1697,7 +1703,8 @@ ts_dimension_add_internal(FunctionCallInfo fcinfo, DimensionInfo *info, bool is_
 															   chunk->fd.id,
 															   slice->fd.id,
 															   NULL,
-															   NULL);
+															   NULL,
+															   info->correlated);
 				ts_chunk_constraint_insert(cc);
 			}
 		}

src/dimension.h

@@ -119,7 +119,8 @@ typedef struct DimensionInfo
 } DimensionInfo;
 
 #define DIMENSION_INFO_IS_SET(di) (di != NULL && OidIsValid((di)->table_relid))
-#define DIMENSION_INFO_IS_VALID(di) (info->num_slices_is_set || OidIsValid(info->interval_type) || info->correlated)
+#define DIMENSION_INFO_IS_VALID(di)                                                                \
+	(info->num_slices_is_set || OidIsValid(info->interval_type) || info->correlated)
 
 extern Hyperspace *ts_dimension_scan(int32 hypertable_id, Oid main_table_relid, int16 num_dimension,
 									 MemoryContext mctx);