Limit ts_array_length to 1 dimensional arrays

This change makes ts_array_length constraints consistant with the
constraints of the other array helper functions. Previously
ts_array_length would also work on multidimensional arrays but by
limiting to 1-dimensional arrays we can optimize the code path.
This patch also adds helper functions for appending an element
to an array.

src/ts_catalog/array_utils.c

@@ -11,21 +11,23 @@
 #include <debug_assert.h>
 #include "array_utils.h"
 
+/*
+ * Array helper function for internal catalog arrays.
+ * These are not suitable for arbitrary dimension
+ * arrays but only for 1-dimensional arrays as we use
+ * them in our catalog.
+ */
+
 extern TSDLLEXPORT int
 ts_array_length(ArrayType *arr)
 {
 	if (!arr)
 		return 0;
 
-	return ArrayGetNItems(ARR_NDIM(arr), ARR_DIMS(arr));
-}
+	Assert(ARR_NDIM(arr) == 1);
 
-/*
- * Array helper function for internal catalog arrays.
- * These are not suitable for arbitrary dimension
- * arrays but only for 1-dimensional arrays as we use
- * them in our catalog.
- */
+	return ARR_DIMS(arr)[0];
+}
 
 extern TSDLLEXPORT bool
 ts_array_is_member(ArrayType *arr, const char *name)
@@ -38,6 +40,7 @@ ts_array_is_member(ArrayType *arr, const char *name)
 
 	Assert(ARR_NDIM(arr) == 1);
 	Assert(arr->elemtype == TEXTOID);
+	Assert(name);
 
 	ArrayIterator it = array_create_iterator(arr, 0, NULL);
 	while (array_iterate(it, &datum, &null))
@@ -126,3 +129,53 @@ ts_array_get_element_text(ArrayType *arr, int position)
 
 	return TextDatumGetCString(value);
 }
+
+extern TSDLLEXPORT ArrayType *
+ts_array_add_element_text(ArrayType *arr, const char *value)
+{
+	Datum val = CStringGetTextDatum(value);
+	if (!arr)
+	{
+		return construct_array(&val, 1, TEXTOID, -1, false, TYPALIGN_INT);
+	}
+	else
+	{
+		Assert(ARR_NDIM(arr) == 1);
+		Assert(arr->elemtype == TEXTOID);
+
+		Datum d = PointerGetDatum(arr);
+
+		int position = ts_array_length(arr);
+		Assert(position);
+		position++;
+
+		d = array_set_element(d, 1, &position, val, false, -1, -1, false, TYPALIGN_INT);
+
+		return DatumGetArrayTypeP(d);
+	}
+}
+
+extern TSDLLEXPORT ArrayType *
+ts_array_add_element_bool(ArrayType *arr, bool value)
+{
+	if (!arr)
+	{
+		Datum val = BoolGetDatum(value);
+		return construct_array(&val, 1, BOOLOID, 1, true, TYPALIGN_CHAR);
+	}
+	else
+	{
+		Assert(ARR_NDIM(arr) == 1);
+		Assert(arr->elemtype == BOOLOID);
+
+		Datum d = PointerGetDatum(arr);
+
+		int position = ts_array_length(arr);
+		Assert(position);
+		position++;
+
+		d = array_set_element(d, 1, &position, value, false, -1, 1, true, TYPALIGN_CHAR);
+
+		return DatumGetArrayTypeP(d);
+	}
+}

src/ts_catalog/array_utils.h

@@ -20,3 +20,5 @@ extern TSDLLEXPORT bool ts_array_is_member(ArrayType *arr, const char *name);
 extern TSDLLEXPORT int ts_array_position(ArrayType *arr, const char *name);
 extern TSDLLEXPORT bool ts_array_get_element_bool(ArrayType *arr, int position);
 extern TSDLLEXPORT const char *ts_array_get_element_text(ArrayType *arr, int position);
+extern TSDLLEXPORT ArrayType *ts_array_add_element_text(ArrayType *arr, const char *value);
+extern TSDLLEXPORT ArrayType *ts_array_add_element_bool(ArrayType *arr, bool value);

tsl/src/chunk_api.c

@@ -44,6 +44,7 @@
 #include "remote/stmt_params.h"
 #include "remote/dist_commands.h"
 #include "remote/tuplefactory.h"
+#include "ts_catalog/array_utils.h"
 #include "ts_catalog/catalog.h"
 #include "ts_catalog/chunk_data_node.h"
 
@@ -1017,7 +1018,7 @@ chunk_update_colstats(Chunk *chunk, int16 attnum, float nullfract, int32 width,
 		Assert(HeapTupleIsValid(type_tuple));
 		type = (Form_pg_type) GETSTRUCT(type_tuple);
 		Assert(slot_values[k] != NULL);
-		nelems = ARR_DIMS(slot_values[k])[0];
+		nelems = ts_array_length(slot_values[k]);
 
 		decoded_data = palloc0(nelems * sizeof(Datum));
 

tsl/src/compression/create.c

@@ -40,6 +40,7 @@
 #include "compression_with_clause.h"
 #include "compression.h"
 #include "hypertable_cache.h"
+#include "ts_catalog/array_utils.h"
 #include "ts_catalog/hypertable_compression.h"
 #include "custom_type_cache.h"
 #include "trigger.h"
@@ -842,7 +843,7 @@ validate_existing_constraints(Hypertable *ht, CompressColInfo *colinfo, List **i
 			}
 
 			arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
-			numkeys = ARR_DIMS(arr)[0];
+			numkeys = ts_array_length(arr);
 			if (ARR_NDIM(arr) != 1 || numkeys < 0 || ARR_HASNULL(arr) ||
 				ARR_ELEMTYPE(arr) != INT2OID)
 				elog(ERROR, "conkey is not a 1-D smallint array");

tsl/src/remote/dist_commands.c

@@ -567,7 +567,7 @@ ts_dist_cmd_exec(PG_FUNCTION_ARGS)
 					 errmsg("invalid data nodes list"),
 					 errdetail("The array of data nodes cannot contain null values.")));
 
-		ndatanodes = ts_array_length(data_nodes);
+		ndatanodes = ArrayGetNItems(ARR_NDIM(data_nodes), ARR_DIMS(data_nodes));
 
 		if (ndatanodes == 0)
 			ereport(ERROR,