text: /*
   * This file and its contents are licensed under the Timescale License.
   * Please see the included NOTICE for copyright information and
   * LICENSE-TIMESCALE for a copy of the license.
   */
  #include <postgres.h>
  #include <executor/spi.h>
  #include <fmgr.h>
  #include <lib/stringinfo.h>
  #include <utils/builtins.h>
  #include <utils/rel.h>
  #include <utils/relcache.h>
  #include <utils/date.h>
  #include <utils/snapmgr.h>
  
  #include <scanner.h>
  #include <compat.h>
  #include <scan_iterator.h>
  #include <continuous_agg.h>
  #include <time_utils.h>
  
  #include "materialize.h"
  
  static bool ranges_overlap(InternalTimeRange invalidation_range,
  						   InternalTimeRange new_materialization_range);
  static TimeRange internal_time_range_to_time_range(InternalTimeRange internal);
  static int64 range_length(const InternalTimeRange range);
  static Datum internal_to_time_value_or_infinite(int64 internal, Oid time_type,
  												bool *is_infinite_out);
  
  /***************************
   * materialization support *
   ***************************/
  
  static void spi_update_materializations(SchemaAndName partial_view,
  										SchemaAndName materialization_table, Name time_column_name,
  										TimeRange invalidation_range);
  static void spi_delete_materializations(SchemaAndName materialization_table, Name time_column_name,
  										TimeRange invalidation_range);
  static void spi_insert_materializations(SchemaAndName partial_view,
  										SchemaAndName materialization_table, Name time_column_name,
  										TimeRange materialization_range);
  
  void
  continuous_agg_update_materialization(SchemaAndName partial_view,
  									  SchemaAndName materialization_table, Name time_column_name,
  									  InternalTimeRange new_materialization_range,
  									  InternalTimeRange invalidation_range, int64 bucket_width)
  {
  	InternalTimeRange combined_materialization_range = new_materialization_range;
  	bool materialize_invalidations_separately = range_length(invalidation_range) > 0;
  	int res = SPI_connect();
  	if (res != SPI_OK_CONNECT)
  		elog(ERROR, "could not connect to SPI in materializer");
  
  	/* pin the start of new_materialization to the end of new_materialization,
  	 * we are not allowed to materialize beyond that point
  	 */
  	if (new_materialization_range.start > new_materialization_range.end)
  		new_materialization_range.start = new_materialization_range.end;
  
  	if (range_length(invalidation_range) > 0)
  	{
  		Assert(invalidation_range.start <= invalidation_range.end);
  
  		/* we never materialize beyond the new materialization range */
  		if (invalidation_range.start >= new_materialization_range.end ||
  			invalidation_range.end > new_materialization_range.end)
  			elog(ERROR, "internal error: invalidation range ahead of new materialization range");
  
  		/* If the invalidation and new materialization ranges overlap, materialize in one go */
  		materialize_invalidations_separately =
  			!ranges_overlap(invalidation_range, new_materialization_range);
  
  		combined_materialization_range.start =
  			int64_min(invalidation_range.start, new_materialization_range.start);
  	}
  
  	/* Then insert the materializations.
  	 * We insert them in two groups:
  	 * [lowest_invalidated, greatest_invalidated] and
  	 * [start_of_new_materialization, end_of_new_materialization]
  	 * eventually, we may want more precise deletions and insertions for the invalidated ranges.
  	 * if greatest_invalidated == end_of_new_materialization then we only perform 1 insertion.
  	 * to prevent values from being inserted multiple times.
  	 */
  	if (range_length(invalidation_range) == 0 || !materialize_invalidations_separately)
  	{
  		spi_update_materializations(partial_view,
  									materialization_table,
  									time_column_name,
  									internal_time_range_to_time_range(
  										combined_materialization_range));
  	}
  	else
  	{
  		spi_update_materializations(partial_view,
  									materialization_table,
  									time_column_name,
  									internal_time_range_to_time_range(invalidation_range));
  
  		spi_update_materializations(partial_view,
  									materialization_table,
  									time_column_name,
  									internal_time_range_to_time_range(new_materialization_range));
  	}
  
  	res = SPI_finish();
  	Assert(res == SPI_OK_FINISH);
  	return;
  }
  
  static bool
  ranges_overlap(InternalTimeRange invalidation_range, InternalTimeRange new_materialization_range)
  {
  	Assert(invalidation_range.start <= invalidation_range.end);
  	Assert(new_materialization_range.start <= new_materialization_range.end);
  	return !(invalidation_range.end < new_materialization_range.start ||
  			 new_materialization_range.end < invalidation_range.start);
  }
  
  static int64
  range_length(const InternalTimeRange range)
  {
  	Assert(range.end >= range.start);
  
  	return int64_saturating_sub(range.end, range.start);
  }
  
  static Datum
  time_range_internal_to_min_time_value(Oid type)
  {
  	switch (type)
  	{
  		case TIMESTAMPOID:
  			return TimestampGetDatum(DT_NOBEGIN);
  		case TIMESTAMPTZOID:
  			return TimestampTzGetDatum(DT_NOBEGIN);
  		case DATEOID:
  			return DateADTGetDatum(DATEVAL_NOBEGIN);
  		default:
  			return ts_internal_to_time_value(PG_INT64_MIN, type);
  	}
  }
  
  static Datum
  time_range_internal_to_max_time_value(Oid type)
  {
  	switch (type)
  	{
  		case TIMESTAMPOID:
  			return TimestampGetDatum(DT_NOEND);
  		case TIMESTAMPTZOID:
  			return TimestampTzGetDatum(DT_NOEND);
  		case DATEOID:
  			return DateADTGetDatum(DATEVAL_NOEND);
  			break;
  		default:
  			return ts_internal_to_time_value(PG_INT64_MAX, type);
  	}
  }
  
  static Datum
  internal_to_time_value_or_infinite(int64 internal, Oid time_type, bool *is_infinite_out)
  {
  	/* MIN and MAX can occur due to NULL thresholds, or due to a lack of invalidations. Since our
  	 * regular conversion function errors in those cases, and we want to use those as markers for an
  	 * open threshold in one direction, we special case this here*/
  	if (internal == PG_INT64_MIN)
  	{
  		if (is_infinite_out != NULL)
  			*is_infinite_out = true;
  		return time_range_internal_to_min_time_value(time_type);
  	}
  	else if (internal == PG_INT64_MAX)
  	{
  		if (is_infinite_out != NULL)
  			*is_infinite_out = true;
  		return time_range_internal_to_max_time_value(time_type);
  	}
  	else
  	{
  		if (is_infinite_out != NULL)
  			*is_infinite_out = false;
  		return ts_internal_to_time_value(internal, time_type);
  	}
  }
  
  static TimeRange
  internal_time_range_to_time_range(InternalTimeRange internal)
  {
  	TimeRange range;
  	range.type = internal.type;
  
  	range.start = internal_to_time_value_or_infinite(internal.start, internal.type, NULL);
  	range.end = internal_to_time_value_or_infinite(internal.end, internal.type, NULL);
  
  	return range;
  }
  
  static void
  spi_update_materializations(SchemaAndName partial_view, SchemaAndName materialization_table,
  							Name time_column_name, TimeRange invalidation_range)
  {
  	spi_delete_materializations(materialization_table, time_column_name, invalidation_range);
  	spi_insert_materializations(partial_view,
  								materialization_table,
  								time_column_name,
  								invalidation_range);
  }
  
  static void
  spi_delete_materializations(SchemaAndName materialization_table, Name time_column_name,
  							TimeRange invalidation_range)
  {
  	int res;
  	StringInfo command = makeStringInfo();
  	Oid out_fn;
  	bool type_is_varlena;
  	char *invalidation_start;
  	char *invalidation_end;
  
  	getTypeOutputInfo(invalidation_range.type, &out_fn, &type_is_varlena);
  	invalidation_start = OidOutputFunctionCall(out_fn, invalidation_range.start);
  	invalidation_end = OidOutputFunctionCall(out_fn, invalidation_range.end);
  
  	appendStringInfo(command,
  					 "DELETE FROM %s.%s AS D WHERE "
  					 "D.%s >= %s AND D.%s < %s;",
  					 quote_identifier(NameStr(*materialization_table.schema)),
  					 quote_identifier(NameStr(*materialization_table.name)),
  					 quote_identifier(NameStr(*time_column_name)),
  					 quote_literal_cstr(invalidation_start),
  					 quote_identifier(NameStr(*time_column_name)),
  					 quote_literal_cstr(invalidation_end));
  
  	res = SPI_execute_with_args(command->data,
  								0 /*=nargs*/,
  								NULL,
  								NULL,
  								NULL /*=Nulls*/,
  								false /*=read_only*/,
  								0 /*count*/);
  	if (res < 0)
  		elog(ERROR, "could not delete old values from materialization table");
  }
  
  static void
  spi_insert_materializations(SchemaAndName partial_view, SchemaAndName materialization_table,
  							Name time_column_name, TimeRange materialization_range)
  {
  	int res;
  	StringInfo command = makeStringInfo();
  	Oid out_fn;
  	bool type_is_varlena;
  	char *materialization_start;
  	char *materialization_end;
  
  	getTypeOutputInfo(materialization_range.type, &out_fn, &type_is_varlena);
  	materialization_start = OidOutputFunctionCall(out_fn, materialization_range.start);
  	materialization_end = OidOutputFunctionCall(out_fn, materialization_range.end);
  
  	appendStringInfo(command,
  					 "INSERT INTO %s.%s SELECT * FROM %s.%s AS I "
  					 "WHERE I.%s >= %s AND I.%s < %s;",
  					 quote_identifier(NameStr(*materialization_table.schema)),
  					 quote_identifier(NameStr(*materialization_table.name)),
  					 quote_identifier(NameStr(*partial_view.schema)),
  					 quote_identifier(NameStr(*partial_view.name)),
  					 quote_identifier(NameStr(*time_column_name)),
  					 quote_literal_cstr(materialization_start),
  					 quote_identifier(NameStr(*time_column_name)),
  					 quote_literal_cstr(materialization_end));
  
  	res = SPI_execute_with_args(command->data,
  								0 /*=nargs*/,
  								NULL /*=argtypes*/,
  								NULL /*=Values*/,
  								NULL /*=Nulls*/,
  								false /*=read_only*/,
  								0 /*count*/);
  
  	if (res < 0)
  		elog(ERROR, "could materialize values into the materialization table");
  }
  