1 /*
2 * PostRR - src/rrtimeslice.c
3 * Copyright (C) 2012 Sebastian 'tokkee' Harl <sh@tokkee.org>
4 * All rights reserved.
5 *
6 * Redistribution and use in source and binary forms, with or without
7 * modification, are permitted provided that the following conditions
8 * are met:
9 * 1. Redistributions of source code must retain the above copyright
10 * notice, this list of conditions and the following disclaimer.
11 * 2. Redistributions in binary form must reproduce the above copyright
12 * notice, this list of conditions and the following disclaimer in the
13 * documentation and/or other materials provided with the distribution.
14 *
15 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
16 * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
17 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
18 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR
19 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
20 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
21 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
22 * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
23 * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
24 * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
25 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
26 */
28 /*
29 * A PostgreSQL data-type providing a timeslice implementing round-robin
30 * features.
31 */
33 #include "postrr.h"
34 #include "utils/pg_spi.h"
36 #include <string.h>
38 #include <postgres.h>
39 #include <fmgr.h>
41 /* Postgres utilities */
42 #include <access/hash.h>
43 #include <executor/spi.h>
44 #include <utils/array.h>
45 #include <utils/datetime.h>
46 #include <utils/timestamp.h>
47 #include <miscadmin.h> /* DateStyle */
49 #ifdef HAVE_INT64_TIMESTAMP
50 # define TSTAMP_TO_INT64(t) (t)
51 # define INT64_TO_TSTAMP(i) (i)
52 #else /* ! HAVE_INT64_TIMESTAMP */
53 # define TSTAMP_TO_INT64(t) (int64)((t) * (double)USECS_PER_SEC)
54 # define INT64_TO_TSTAMP(i) ((double)(i) / (double)USECS_PER_SEC)
55 #endif
57 /*
58 * data type
59 */
61 struct rrtimeslice {
62 TimestampTz tstamp;
63 int32 tsid;
64 uint32 seq;
65 };
67 /*
68 * internal helper functions
69 */
71 static int32
72 rrtimeslice_set_spec(int32 len, int32 num)
73 {
74 int spi_rc;
76 char query[256];
77 int32 typmod = 0;
79 if ((len <= 0) || (num <= 0))
80 ereport(ERROR, (
81 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
82 errmsg("rrtimeslice(%i, %i) "
83 "length/num may not be less than zero",
84 len, num)
85 ));
87 if ((spi_rc = SPI_connect()) != SPI_OK_CONNECT)
88 ereport(ERROR, (
89 errmsg("failed to store rrtimeslice spec: "
90 "could not connect to SPI manager: %s",
91 SPI_result_code_string(spi_rc))
92 ));
94 snprintf(query, sizeof(query),
95 "SELECT tsid FROM postrr.rrtimeslices "
96 "WHERE tslen = %d AND tsnum = %d "
97 "LIMIT 1", len, num);
98 query[sizeof(query) - 1] = '\0';
100 spi_rc = pg_spi_get_int(query, 1, &typmod);
101 if (spi_rc == PG_SPI_OK) {
102 SPI_finish();
103 return typmod;
104 }
105 else if (spi_rc != PG_SPI_ERROR_NO_VALUES)
106 pg_spi_ereport(ERROR, "store rrtimeslice spec", spi_rc);
108 snprintf(query, sizeof(query),
109 "SELECT nextval('postrr.tsid'::regclass)");
110 query[sizeof(query) - 1] = '\0';
112 spi_rc = pg_spi_get_int(query, 1, &typmod);
113 if ((spi_rc != PG_SPI_OK) || (typmod <= 0))
114 pg_spi_ereport(ERROR, "retrieve nextval(postrr.tsid)", spi_rc);
116 snprintf(query, sizeof(query),
117 "INSERT INTO postrr.rrtimeslices(tsid, tslen, tsnum) "
118 "VALUES (%d, %d, %d)", typmod, len, num);
119 query[sizeof(query) - 1] = '\0';
121 spi_rc = SPI_exec(query, /* max num rows = */ 1);
122 if (spi_rc != SPI_OK_INSERT)
123 ereport(ERROR, (
124 errmsg("failed to store rrtimeslice spec: "
125 "failed to execute query: %s",
126 SPI_result_code_string(spi_rc))
127 ));
129 SPI_finish();
130 return typmod;
131 } /* rrtimeslice_set_spec */
133 static int
134 rrtimeslice_get_spec(int32 typmod, int32 *len, int32 *num)
135 {
136 int spi_rc;
138 char query[256];
140 if (typmod <= 0)
141 return -1;
143 if ((spi_rc = SPI_connect()) != SPI_OK_CONNECT)
144 ereport(ERROR, (
145 errmsg("failed to determine rrtimeslice spec: "
146 "could not connect to SPI manager: %s",
147 SPI_result_code_string(spi_rc))
148 ));
150 snprintf(query, sizeof(query),
151 "SELECT tslen, tsnum FROM postrr.rrtimeslices "
152 "WHERE tsid = %d", typmod);
153 query[sizeof(query) - 1] = '\0';
155 spi_rc = pg_spi_get_int(query, 2, len, num);
156 if (spi_rc != PG_SPI_OK)
157 pg_spi_ereport(ERROR, "determine rrtimeslice spec", spi_rc);
159 SPI_finish();
160 return 0;
161 } /* rrtimeslice_get_spec */
163 static int
164 rrtimeslice_apply_typmod(rrtimeslice_t *tslice, int32 typmod)
165 {
166 int64 tstamp;
167 int64 length;
168 int64 seq;
170 int32 len = 0;
171 int32 num = 0;
173 if (rrtimeslice_get_spec(typmod, &len, &num))
174 return -1;
176 if ((len <= 0) || (num <= 0))
177 ereport(ERROR, (
178 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
179 errmsg("rrtimeslice(%i, %i) "
180 "length/num may not be less than zero",
181 len, num)
182 ));
184 tstamp = TSTAMP_TO_INT64(tslice->tstamp);
186 length = len * USECS_PER_SEC;
187 if (tstamp % length != 0)
188 tstamp = tstamp - (tstamp % length) + length;
189 seq = tstamp % (length * num) / length;
190 seq = seq % num;
192 tslice->tstamp = INT64_TO_TSTAMP(tstamp);
193 tslice->tsid = typmod;
194 tslice->seq = (uint32)seq;
195 return 0;
196 } /* rrtimeslice_apply_typmod */
198 /*
199 * prototypes for PostgreSQL functions
200 */
202 PG_FUNCTION_INFO_V1(rrtimeslice_validate);
204 PG_FUNCTION_INFO_V1(rrtimeslice_in);
205 PG_FUNCTION_INFO_V1(rrtimeslice_out);
206 PG_FUNCTION_INFO_V1(rrtimeslice_typmodin);
207 PG_FUNCTION_INFO_V1(rrtimeslice_typmodout);
209 PG_FUNCTION_INFO_V1(rrtimeslice_to_rrtimeslice);
210 PG_FUNCTION_INFO_V1(rrtimeslice_to_timestamptz);
212 PG_FUNCTION_INFO_V1(rrtimeslice_seq_eq);
213 PG_FUNCTION_INFO_V1(rrtimeslice_seq_ne);
214 PG_FUNCTION_INFO_V1(rrtimeslice_seq_lt);
215 PG_FUNCTION_INFO_V1(rrtimeslice_seq_gt);
216 PG_FUNCTION_INFO_V1(rrtimeslice_seq_le);
217 PG_FUNCTION_INFO_V1(rrtimeslice_seq_ge);
218 PG_FUNCTION_INFO_V1(rrtimeslice_seq_cmp);
219 PG_FUNCTION_INFO_V1(rrtimeslice_seq_hash);
221 /*
222 * public API
223 */
225 Datum
226 rrtimeslice_validate(PG_FUNCTION_ARGS)
227 {
228 char type_info[1024];
229 char *result;
230 size_t req_len;
231 size_t len;
233 if (PG_NARGS() != 1)
234 ereport(ERROR, (
235 errmsg("rrtimeslice_validate() expect one argument"),
236 errhint("Usage rrtimeslice_validate(expected_size)")
237 ));
239 req_len = (size_t)PG_GETARG_UINT32(0);
240 len = sizeof(rrtimeslice_t);
242 if (req_len != len)
243 ereport(ERROR, (
244 errmsg("length of the rrtimeslice type "
245 "does not match the expected length"),
246 errhint("Please report a bug against PostRR")
247 ));
249 snprintf(type_info, sizeof(type_info),
250 "rrtimeslice validated successfully; type length = %zu", len);
251 type_info[sizeof(type_info) - 1] = '\0';
253 result = pstrdup(type_info);
254 PG_RETURN_CSTRING(result);
255 } /* rrtimeslice_validate */
257 Datum
258 rrtimeslice_in(PG_FUNCTION_ARGS)
259 {
260 rrtimeslice_t *tslice;
262 TimestampTz tstamp = 0;
263 int32 typmod;
265 struct pg_tm tm;
266 fsec_t fsec = 0;
267 int tz = 0;
269 char *time_str;
270 int pg_dt_err;
271 char buf[MAXDATELEN + MAXDATEFIELDS];
272 char *field[MAXDATEFIELDS];
273 int ftype[MAXDATEFIELDS];
274 int num_fields = 0;
275 int dtype = 0;
277 if (PG_NARGS() != 3)
278 ereport(ERROR, (
279 errmsg("rrtimeslice_in() expects three arguments"),
280 errhint("Usage: rrtimeslice_in(col_name, oid, typmod)")
281 ));
283 tslice = (rrtimeslice_t *)palloc0(sizeof(*tslice));
284 time_str = PG_GETARG_CSTRING(0);
285 typmod = PG_GETARG_INT32(2);
287 pg_dt_err = ParseDateTime(time_str, buf, sizeof(buf),
288 field, ftype, MAXDATEFIELDS, &num_fields);
290 if (! pg_dt_err)
291 pg_dt_err = DecodeDateTime(field, ftype, num_fields,
292 &dtype, &tm, &fsec, &tz);
293 if (pg_dt_err)
294 DateTimeParseError(pg_dt_err, time_str, "rrtimeslice");
296 switch (dtype) {
297 case DTK_DATE:
298 if (tm2timestamp(&tm, fsec, &tz, &tstamp))
299 ereport(ERROR, (
300 errcode(ERRCODE_DATETIME_VALUE_OUT_OF_RANGE),
301 errmsg("timestamp out of range: %s", time_str)
302 ));
303 break;
305 case DTK_EPOCH:
306 tstamp = SetEpochTimestamp();
307 break;
309 default:
310 ereport(ERROR, (
311 errmsg("unexpected dtype %d while "
312 "parsing rrtimeslice: %s", dtype, time_str)
313 ));
314 }
316 tslice->tstamp = tstamp;
318 /* most likely, this won't happen … coerce_type
319 * (src/backend/parser/parse_coerce.c) does not pass that information to
320 * the input function but rather lets a length conversion cast do that
321 * work */
322 if (typmod > 0)
323 rrtimeslice_apply_typmod(tslice, typmod);
325 PG_RETURN_RRTIMESLICE_P(tslice);
326 } /* rrtimeslice_in */
328 Datum
329 rrtimeslice_out(PG_FUNCTION_ARGS)
330 {
331 rrtimeslice_t *tslice;
333 struct pg_tm tm;
334 fsec_t fsec = 0;
335 int tz = 0;
337 char *tz_str = NULL;
339 char ts_str[MAXDATELEN + 1];
340 char buf_l[MAXDATELEN + 1];
341 char buf_u[MAXDATELEN + 1];
342 char *result;
344 int32 len = 0;
345 int32 num = 0;
347 if (PG_NARGS() != 1)
348 ereport(ERROR, (
349 errmsg("rrtimeslice_out() expects one argument"),
350 errhint("Usage: rrtimeslice_out(rrtimeslice)")
351 ));
353 tslice = PG_GETARG_RRTIMESLICE_P(0);
355 if (TIMESTAMP_NOT_FINITE(tslice->tstamp)
356 || timestamp2tm(tslice->tstamp, &tz, &tm, &fsec, &tz_str, NULL))
357 ereport(ERROR, (
358 errcode(ERRCODE_DATETIME_VALUE_OUT_OF_RANGE),
359 errmsg("invalid (non-finite) timestamp")
360 ));
362 EncodeDateTime(&tm, fsec, &tz, &tz_str, DateStyle, buf_u);
364 if (! rrtimeslice_get_spec(tslice->tsid, &len, &num)) {
365 TimestampTz lower = tslice->tstamp - (len * USECS_PER_SEC);
367 if (timestamp2tm(lower, &tz, &tm, &fsec, &tz_str, NULL))
368 ereport(ERROR, (
369 errcode(ERRCODE_DATETIME_VALUE_OUT_OF_RANGE),
370 errmsg("invalid (non-finite) lower timestamp")
371 ));
373 EncodeDateTime(&tm, fsec, &tz, &tz_str, DateStyle, buf_l);
374 }
375 else {
376 strncpy(buf_l, "ERR", sizeof(buf_l));
377 buf_l[sizeof(buf_l) - 1] = '\0';
378 }
380 snprintf(ts_str, sizeof(ts_str), "(\"%s\", \"%s\"] #%i/%i",
381 buf_l, buf_u, tslice->seq, num);
383 result = pstrdup(ts_str);
384 PG_RETURN_CSTRING(result);
385 } /* rrtimeslice_out */
387 Datum
388 rrtimeslice_typmodin(PG_FUNCTION_ARGS)
389 {
390 ArrayType *tm_array;
392 int32 *spec;
393 int spec_elems = 0;
394 int32 typmod;
396 if (PG_NARGS() != 1)
397 ereport(ERROR, (
398 errmsg("rrtimeslice_typmodin() expects one argument"),
399 errhint("Usage: rrtimeslice_typmodin(array)")
400 ));
402 tm_array = PG_GETARG_ARRAYTYPE_P(0);
404 spec = ArrayGetIntegerTypmods(tm_array, &spec_elems);
405 if (spec_elems != 2)
406 ereport(ERROR, (
407 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
408 errmsg("invalid rrtimeslice type modifier"),
409 errhint("Usage: rrtimeslice(<slice_len>, <num>)")
410 ));
412 typmod = rrtimeslice_set_spec(spec[0], spec[1]);
413 PG_RETURN_INT32(typmod);
414 } /* rrtimeslice_typmodin */
416 Datum
417 rrtimeslice_typmodout(PG_FUNCTION_ARGS)
418 {
419 int32 typmod;
420 char tm_str[1024];
421 char *result;
423 int32 len = 0;
424 int32 num = 0;
426 if (PG_NARGS() != 1)
427 ereport(ERROR, (
428 errmsg("rrtimeslice_typmodout() expects one argument"),
429 errhint("Usage: rrtimeslice_typmodout(typmod)")
430 ));
432 typmod = PG_GETARG_INT32(0);
433 if (rrtimeslice_get_spec(typmod, &len, &num))
434 tm_str[0] = '\0';
435 else if ((len <= 0) || (num <= 0))
436 snprintf(tm_str, sizeof(tm_str), "(#ERR, #ERR)");
437 else
438 snprintf(tm_str, sizeof(tm_str), "(%d, %d)", len, num);
440 result = pstrdup(tm_str);
441 PG_RETURN_CSTRING(result);
442 } /* rrtimeslice_typmodout */
444 Datum
445 rrtimeslice_to_rrtimeslice(PG_FUNCTION_ARGS)
446 {
447 rrtimeslice_t *tslice;
448 int32 typmod;
450 if (PG_NARGS() != 3)
451 ereport(ERROR, (
452 errmsg("rrtimeslice_to_rrtimeslice() "
453 "expects three arguments"),
454 errhint("Usage: rrtimeslice_to_rrtimeslice"
455 "(rrtimeslice, typmod, is_explicit)")
456 ));
458 tslice = PG_GETARG_RRTIMESLICE_P(0);
459 typmod = PG_GETARG_INT32(1);
461 if (typmod > 0) {
462 if ((! tslice->tsid) && (! tslice->seq))
463 rrtimeslice_apply_typmod(tslice, typmod);
464 else
465 ereport(ERROR, (
466 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
467 errmsg("invalid cast: cannot cast rrtimeslices "
468 "with different typmod (yet)")
469 ));
470 }
472 PG_RETURN_RRTIMESLICE_P(tslice);
473 } /* rrtimeslice_to_rrtimeslice */
475 Datum
476 rrtimeslice_to_timestamptz(PG_FUNCTION_ARGS)
477 {
478 rrtimeslice_t *tslice;
480 if (PG_NARGS() != 1)
481 ereport(ERROR, (
482 errmsg("rrtimeslice_to_timestamptz() "
483 "expects one argument"),
484 errhint("Usage: rrtimeslice_to_timestamptz"
485 "(rrtimeslice)")
486 ));
488 tslice = PG_GETARG_RRTIMESLICE_P(0);
489 PG_RETURN_TIMESTAMPTZ(tslice->tstamp);
490 } /* rrtimeslice_to_timestamptz */
492 int
493 rrtimeslice_seq_cmp_internal(rrtimeslice_t *ts1, rrtimeslice_t *ts2)
494 {
495 if ((! ts1) && (! ts2))
496 return 0;
497 else if (! ts1)
498 return -1;
499 else if (! ts2)
500 return 1;
502 if (ts1->tsid && (! ts2->tsid))
503 rrtimeslice_apply_typmod(ts2, ts1->tsid);
504 else if ((! ts1->tsid) && ts2->tsid)
505 rrtimeslice_apply_typmod(ts1, ts2->tsid);
507 if (ts1->tsid != ts2->tsid) /* XXX: compare len/num */
508 ereport(ERROR, (
509 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
510 errmsg("invalid comparison: cannot compare "
511 "rrtimeslices with different typmods (yet)")
512 ));
514 if (ts1->seq < ts2->seq)
515 return -1;
516 else if (ts1->seq == ts2->seq)
517 return 0;
518 else
519 return 1;
520 } /* rrtimeslice_seq_cmp_internal */
522 Datum
523 rrtimeslice_seq_eq(PG_FUNCTION_ARGS)
524 {
525 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
526 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
528 PG_RETURN_BOOL(rrtimeslice_seq_cmp_internal(ts1, ts2) == 0);
529 } /* rrtimeslice_seq_eq */
531 Datum
532 rrtimeslice_seq_ne(PG_FUNCTION_ARGS)
533 {
534 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
535 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
537 PG_RETURN_BOOL(rrtimeslice_seq_cmp_internal(ts1, ts2) != 0);
538 } /* rrtimeslice_seq_ne */
540 Datum
541 rrtimeslice_seq_lt(PG_FUNCTION_ARGS)
542 {
543 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
544 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
546 PG_RETURN_BOOL(rrtimeslice_seq_cmp_internal(ts1, ts2) < 0);
547 } /* rrtimeslice_seq_lt */
549 Datum
550 rrtimeslice_seq_le(PG_FUNCTION_ARGS)
551 {
552 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
553 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
555 PG_RETURN_BOOL(rrtimeslice_seq_cmp_internal(ts1, ts2) <= 0);
556 } /* rrtimeslice_seq_le */
558 Datum
559 rrtimeslice_seq_gt(PG_FUNCTION_ARGS)
560 {
561 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
562 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
564 PG_RETURN_BOOL(rrtimeslice_seq_cmp_internal(ts1, ts2) > 0);
565 } /* rrtimeslice_seq_gt */
567 Datum
568 rrtimeslice_seq_ge(PG_FUNCTION_ARGS)
569 {
570 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
571 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
573 PG_RETURN_BOOL(rrtimeslice_seq_cmp_internal(ts1, ts2) >= 0);
574 } /* rrtimeslice_seq_ge */
576 Datum
577 rrtimeslice_seq_cmp(PG_FUNCTION_ARGS)
578 {
579 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
580 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
582 PG_RETURN_INT32(rrtimeslice_seq_cmp_internal(ts1, ts2));
583 } /* rrtimeslice_seq_ge */
585 Datum
586 rrtimeslice_seq_hash(PG_FUNCTION_ARGS)
587 {
588 rrtimeslice_t *ts = PG_GETARG_RRTIMESLICE_P(0);
589 return hash_uint32(ts->seq);
590 } /* rrtimeslice_seq_hash */
592 /* vim: set tw=78 sw=4 ts=4 noexpandtab : */