4f151d27b1b3f93b60164bc9759ddfb718396b46
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, IntervalStyle */
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_ts[MAXDATELEN + 1];
341 char buf_l[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) != 0))
357 ereport(ERROR, (
358 errcode(ERRCODE_DATETIME_VALUE_OUT_OF_RANGE),
359 errmsg("invalid (non-finite) timestamp")
360 ));
361 else
362 EncodeDateTime(&tm, fsec, &tz, &tz_str, DateStyle, buf_ts);
364 if (! rrtimeslice_get_spec(tslice->tsid, &len, &num)) {
365 Interval interval;
367 memset(&interval, 0, sizeof(interval));
368 interval.time = len * USECS_PER_SEC;
369 fsec = 0;
371 if (interval2tm(interval, &tm, &fsec))
372 ereport(ERROR, (
373 errmsg("could not convert interval to tm")
374 ));
376 EncodeInterval(&tm, fsec, IntervalStyle, buf_l);
377 }
378 else {
379 strncpy(buf_l, "ERR", sizeof(buf_l));
380 buf_l[sizeof(buf_l) - 1] = '\0';
381 }
383 snprintf(ts_str, sizeof(ts_str), "%s -%s (#%i)",
384 buf_ts, buf_l, tslice->seq);
386 result = pstrdup(ts_str);
387 PG_RETURN_CSTRING(result);
388 } /* rrtimeslice_out */
390 Datum
391 rrtimeslice_typmodin(PG_FUNCTION_ARGS)
392 {
393 ArrayType *tm_array;
395 int32 *spec;
396 int spec_elems = 0;
397 int32 typmod;
399 if (PG_NARGS() != 1)
400 ereport(ERROR, (
401 errmsg("rrtimeslice_typmodin() expects one argument"),
402 errhint("Usage: rrtimeslice_typmodin(array)")
403 ));
405 tm_array = PG_GETARG_ARRAYTYPE_P(0);
407 spec = ArrayGetIntegerTypmods(tm_array, &spec_elems);
408 if (spec_elems != 2)
409 ereport(ERROR, (
410 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
411 errmsg("invalid rrtimeslice type modifier"),
412 errhint("Usage: rrtimeslice(<slice_len>, <num>)")
413 ));
415 typmod = rrtimeslice_set_spec(spec[0], spec[1]);
416 PG_RETURN_INT32(typmod);
417 } /* rrtimeslice_typmodin */
419 Datum
420 rrtimeslice_typmodout(PG_FUNCTION_ARGS)
421 {
422 int32 typmod;
423 char tm_str[1024];
424 char *result;
426 int32 len = 0;
427 int32 num = 0;
429 if (PG_NARGS() != 1)
430 ereport(ERROR, (
431 errmsg("rrtimeslice_typmodout() expects one argument"),
432 errhint("Usage: rrtimeslice_typmodout(typmod)")
433 ));
435 typmod = PG_GETARG_INT32(0);
436 if (rrtimeslice_get_spec(typmod, &len, &num))
437 tm_str[0] = '\0';
438 else if ((len <= 0) || (num <= 0))
439 snprintf(tm_str, sizeof(tm_str), "(#ERR, #ERR)");
440 else
441 snprintf(tm_str, sizeof(tm_str), "(%d, %d)", len, num);
443 result = pstrdup(tm_str);
444 PG_RETURN_CSTRING(result);
445 } /* rrtimeslice_typmodout */
447 Datum
448 rrtimeslice_to_rrtimeslice(PG_FUNCTION_ARGS)
449 {
450 rrtimeslice_t *tslice;
451 int32 typmod;
453 if (PG_NARGS() != 3)
454 ereport(ERROR, (
455 errmsg("rrtimeslice_to_rrtimeslice() "
456 "expects three arguments"),
457 errhint("Usage: rrtimeslice_to_rrtimeslice"
458 "(rrtimeslice, typmod, is_explicit)")
459 ));
461 tslice = PG_GETARG_RRTIMESLICE_P(0);
462 typmod = PG_GETARG_INT32(1);
464 if (typmod > 0) {
465 if ((! tslice->tsid) && (! tslice->seq))
466 rrtimeslice_apply_typmod(tslice, typmod);
467 else
468 ereport(ERROR, (
469 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
470 errmsg("invalid cast: cannot cast rrtimeslices "
471 "with different typmod (yet)")
472 ));
473 }
475 PG_RETURN_RRTIMESLICE_P(tslice);
476 } /* rrtimeslice_to_rrtimeslice */
478 Datum
479 rrtimeslice_to_timestamptz(PG_FUNCTION_ARGS)
480 {
481 rrtimeslice_t *tslice;
483 if (PG_NARGS() != 1)
484 ereport(ERROR, (
485 errmsg("rrtimeslice_to_timestamptz() "
486 "expects one argument"),
487 errhint("Usage: rrtimeslice_to_timestamptz"
488 "(rrtimeslice)")
489 ));
491 tslice = PG_GETARG_RRTIMESLICE_P(0);
492 PG_RETURN_TIMESTAMPTZ(tslice->tstamp);
493 } /* rrtimeslice_to_timestamptz */
495 int
496 rrtimeslice_seq_cmp_internal(rrtimeslice_t *ts1, rrtimeslice_t *ts2)
497 {
498 if ((! ts1) && (! ts2))
499 return 0;
500 else if (! ts1)
501 return -1;
502 else if (! ts2)
503 return 1;
505 if (ts1->tsid && (! ts2->tsid))
506 rrtimeslice_apply_typmod(ts2, ts1->tsid);
507 else if ((! ts1->tsid) && ts2->tsid)
508 rrtimeslice_apply_typmod(ts1, ts2->tsid);
510 if (ts1->tsid != ts2->tsid) /* XXX: compare len/num */
511 ereport(ERROR, (
512 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
513 errmsg("invalid comparison: cannot compare "
514 "rrtimeslices with different typmods (yet)")
515 ));
517 if (ts1->seq < ts2->seq)
518 return -1;
519 else if (ts1->seq == ts2->seq)
520 return 0;
521 else
522 return 1;
523 } /* rrtimeslice_seq_cmp_internal */
525 Datum
526 rrtimeslice_seq_eq(PG_FUNCTION_ARGS)
527 {
528 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
529 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
531 PG_RETURN_BOOL(rrtimeslice_seq_cmp_internal(ts1, ts2) == 0);
532 } /* rrtimeslice_seq_eq */
534 Datum
535 rrtimeslice_seq_ne(PG_FUNCTION_ARGS)
536 {
537 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
538 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
540 PG_RETURN_BOOL(rrtimeslice_seq_cmp_internal(ts1, ts2) != 0);
541 } /* rrtimeslice_seq_ne */
543 Datum
544 rrtimeslice_seq_lt(PG_FUNCTION_ARGS)
545 {
546 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
547 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
549 PG_RETURN_BOOL(rrtimeslice_seq_cmp_internal(ts1, ts2) < 0);
550 } /* rrtimeslice_seq_lt */
552 Datum
553 rrtimeslice_seq_le(PG_FUNCTION_ARGS)
554 {
555 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
556 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
558 PG_RETURN_BOOL(rrtimeslice_seq_cmp_internal(ts1, ts2) <= 0);
559 } /* rrtimeslice_seq_le */
561 Datum
562 rrtimeslice_seq_gt(PG_FUNCTION_ARGS)
563 {
564 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
565 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
567 PG_RETURN_BOOL(rrtimeslice_seq_cmp_internal(ts1, ts2) > 0);
568 } /* rrtimeslice_seq_gt */
570 Datum
571 rrtimeslice_seq_ge(PG_FUNCTION_ARGS)
572 {
573 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
574 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
576 PG_RETURN_BOOL(rrtimeslice_seq_cmp_internal(ts1, ts2) >= 0);
577 } /* rrtimeslice_seq_ge */
579 Datum
580 rrtimeslice_seq_cmp(PG_FUNCTION_ARGS)
581 {
582 rrtimeslice_t *ts1 = PG_GETARG_RRTIMESLICE_P(0);
583 rrtimeslice_t *ts2 = PG_GETARG_RRTIMESLICE_P(1);
585 PG_RETURN_INT32(rrtimeslice_seq_cmp_internal(ts1, ts2));
586 } /* rrtimeslice_seq_ge */
588 Datum
589 rrtimeslice_seq_hash(PG_FUNCTION_ARGS)
590 {
591 rrtimeslice_t *ts = PG_GETARG_RRTIMESLICE_P(0);
592 return hash_uint32(ts->seq);
593 } /* rrtimeslice_seq_hash */
595 /* vim: set tw=78 sw=4 ts=4 noexpandtab : */