-
Notifications
You must be signed in to change notification settings - Fork 35
/
Pg.pm
4517 lines (3432 loc) · 189 KB
/
Pg.pm
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# -*-cperl-*-
#
# Copyright (c) 2002-2024 Greg Sabino Mullane and others: see the Changes file
# Portions Copyright (c) 2002 Jeffrey W. Baker
# Portions Copyright (c) 1997-2001 Edmund Mergl
# Portions Copyright (c) 1994-1997 Tim Bunce
#
# You may distribute under the terms of either the GNU General Public
# License or the Artistic License, as specified in the Perl README file.
use strict;
use warnings;
use 5.008001;
{
package DBD::Pg;
use version; our $VERSION = qv('3.18.0');
use DBI 1.614 ();
use Exporter ();
use XSLoader;
our $dbh;
our @ISA = qw(Exporter);
use constant {
PG_MIN_SMALLINT => -32768,
PG_MAX_SMALLINT => 32767,
PG_MIN_INTEGER => -2147483648,
PG_MAX_INTEGER => 2147483647,
PG_MIN_BIGINT => '-9223372036854775808',
PG_MAX_BIGINT => '9223372036854775807',
PG_MIN_SMALLSERIAL => 1,
PG_MAX_SMALLSERIAL => 32767,
PG_MIN_SERIAL => 1,
PG_MAX_SERIAL => 2147483647,
PG_MIN_BIGSERIAL => 1,
PG_MAX_BIGSERIAL => '9223372036854775807',
};
our %EXPORT_TAGS =
(
async => [qw($DBDPG_DEFAULT PG_ASYNC PG_OLDQUERY_CANCEL PG_OLDQUERY_WAIT)],
pg_limits => [qw($DBDPG_DEFAULT
PG_MIN_SMALLINT PG_MAX_SMALLINT PG_MIN_INTEGER PG_MAX_INTEGER PG_MAX_BIGINT PG_MIN_BIGINT
PG_MIN_SMALLSERIAL PG_MAX_SMALLSERIAL PG_MIN_SERIAL PG_MAX_SERIAL PG_MIN_BIGSERIAL PG_MAX_BIGSERIAL)],
pg_types => [qw($DBDPG_DEFAULT PG_ASYNC PG_OLDQUERY_CANCEL PG_OLDQUERY_WAIT
PG_ACLITEM PG_ACLITEMARRAY PG_ANY PG_ANYARRAY PG_ANYCOMPATIBLE
PG_ANYCOMPATIBLEARRAY PG_ANYCOMPATIBLEMULTIRANGE PG_ANYCOMPATIBLENONARRAY PG_ANYCOMPATIBLERANGE PG_ANYELEMENT
PG_ANYENUM PG_ANYMULTIRANGE PG_ANYNONARRAY PG_ANYRANGE PG_BIT
PG_BITARRAY PG_BOOL PG_BOOLARRAY PG_BOX PG_BOXARRAY
PG_BPCHAR PG_BPCHARARRAY PG_BYTEA PG_BYTEAARRAY PG_CHAR
PG_CHARARRAY PG_CID PG_CIDARRAY PG_CIDR PG_CIDRARRAY
PG_CIRCLE PG_CIRCLEARRAY PG_CSTRING PG_CSTRINGARRAY PG_DATE
PG_DATEARRAY PG_DATEMULTIRANGE PG_DATEMULTIRANGEARRAY PG_DATERANGE PG_DATERANGEARRAY
PG_EVENT_TRIGGER PG_FDW_HANDLER PG_FLOAT4 PG_FLOAT4ARRAY PG_FLOAT8
PG_FLOAT8ARRAY PG_GTSVECTOR PG_GTSVECTORARRAY PG_INDEX_AM_HANDLER PG_INET
PG_INETARRAY PG_INT2 PG_INT2ARRAY PG_INT2VECTOR PG_INT2VECTORARRAY
PG_INT4 PG_INT4ARRAY PG_INT4MULTIRANGE PG_INT4MULTIRANGEARRAY PG_INT4RANGE
PG_INT4RANGEARRAY PG_INT8 PG_INT8ARRAY PG_INT8MULTIRANGE PG_INT8MULTIRANGEARRAY
PG_INT8RANGE PG_INT8RANGEARRAY PG_INTERNAL PG_INTERVAL PG_INTERVALARRAY
PG_JSON PG_JSONARRAY PG_JSONB PG_JSONBARRAY PG_JSONPATH
PG_JSONPATHARRAY PG_LANGUAGE_HANDLER PG_LINE PG_LINEARRAY PG_LSEG
PG_LSEGARRAY PG_MACADDR PG_MACADDR8 PG_MACADDR8ARRAY PG_MACADDRARRAY
PG_MONEY PG_MONEYARRAY PG_NAME PG_NAMEARRAY PG_NUMERIC
PG_NUMERICARRAY PG_NUMMULTIRANGE PG_NUMMULTIRANGEARRAY PG_NUMRANGE PG_NUMRANGEARRAY
PG_OID PG_OIDARRAY PG_OIDVECTOR PG_OIDVECTORARRAY PG_PATH
PG_PATHARRAY PG_PG_ATTRIBUTE PG_PG_ATTRIBUTEARRAY PG_PG_BRIN_BLOOM_SUMMARY PG_PG_BRIN_MINMAX_MULTI_SUMMARY
PG_PG_CLASS PG_PG_CLASSARRAY PG_PG_DDL_COMMAND PG_PG_DEPENDENCIES PG_PG_LSN
PG_PG_LSNARRAY PG_PG_MCV_LIST PG_PG_NDISTINCT PG_PG_NODE_TREE PG_PG_PROC
PG_PG_PROCARRAY PG_PG_SNAPSHOT PG_PG_SNAPSHOTARRAY PG_PG_TYPE PG_PG_TYPEARRAY
PG_POINT PG_POINTARRAY PG_POLYGON PG_POLYGONARRAY PG_RECORD
PG_RECORDARRAY PG_REFCURSOR PG_REFCURSORARRAY PG_REGCLASS PG_REGCLASSARRAY
PG_REGCOLLATION PG_REGCOLLATIONARRAY PG_REGCONFIG PG_REGCONFIGARRAY PG_REGDICTIONARY
PG_REGDICTIONARYARRAY PG_REGNAMESPACE PG_REGNAMESPACEARRAY PG_REGOPER PG_REGOPERARRAY
PG_REGOPERATOR PG_REGOPERATORARRAY PG_REGPROC PG_REGPROCARRAY PG_REGPROCEDURE
PG_REGPROCEDUREARRAY PG_REGROLE PG_REGROLEARRAY PG_REGTYPE PG_REGTYPEARRAY
PG_TABLE_AM_HANDLER PG_TEXT PG_TEXTARRAY PG_TID PG_TIDARRAY
PG_TIME PG_TIMEARRAY PG_TIMESTAMP PG_TIMESTAMPARRAY PG_TIMESTAMPTZ
PG_TIMESTAMPTZARRAY PG_TIMETZ PG_TIMETZARRAY PG_TRIGGER PG_TSMULTIRANGE
PG_TSMULTIRANGEARRAY PG_TSM_HANDLER PG_TSQUERY PG_TSQUERYARRAY PG_TSRANGE
PG_TSRANGEARRAY PG_TSTZMULTIRANGE PG_TSTZMULTIRANGEARRAY PG_TSTZRANGE PG_TSTZRANGEARRAY
PG_TSVECTOR PG_TSVECTORARRAY PG_TXID_SNAPSHOT PG_TXID_SNAPSHOTARRAY PG_UNKNOWN
PG_UUID PG_UUIDARRAY PG_VARBIT PG_VARBITARRAY PG_VARCHAR
PG_VARCHARARRAY PG_VOID PG_XID PG_XID8 PG_XID8ARRAY
PG_XIDARRAY PG_XML PG_XMLARRAY
)],
);
{
package DBD::Pg::DefaultValue;
sub new { my $self = {}; return bless $self, shift; }
}
our $DBDPG_DEFAULT = DBD::Pg::DefaultValue->new();
Exporter::export_ok_tags('pg_types', 'async', 'pg_limits');
our @EXPORT = qw($DBDPG_DEFAULT PG_ASYNC PG_OLDQUERY_CANCEL PG_OLDQUERY_WAIT PG_BYTEA);
XSLoader::load(__PACKAGE__, $VERSION);
our $err = 0; # holds error code for DBI::err
our $errstr = ''; # holds error string for DBI::errstr
our $sqlstate = ''; # holds five character SQLSTATE code
our $drh = undef; # holds driver handle once initialized
## These two methods are here to allow calling before connect()
sub parse_trace_flag {
my ($class, $flag) = @_;
return (0x7FFFFF00 - 0x08000000) if $flag eq 'DBD'; ## all but the prefix
return 0x01000000 if $flag eq 'pglibpq';
return 0x02000000 if $flag eq 'pgstart';
return 0x04000000 if $flag eq 'pgend';
return 0x08000000 if $flag eq 'pgprefix';
return 0x10000000 if $flag eq 'pglogin';
return 0x20000000 if $flag eq 'pgquote';
return DBI::parse_trace_flag($class, $flag);
}
sub parse_trace_flags {
my ($class, $flags) = @_;
return DBI::parse_trace_flags($class, $flags);
}
## Both CLONE and driver are required by DBI, see perldoc DBI::DBD
sub CLONE {
$drh = undef;
return;
}
my $methods_are_installed = 0;
sub driver {
return $drh if defined $drh;
my $class = shift;
$class .= '::dr';
## Work around for issue found in https://rt.cpan.org/Ticket/Display.html?id=83057
my $realversion = qv('3.18.0');
$drh = DBI::_new_drh($class, {
'Name' => 'Pg',
'Version' => $realversion,
'Err' => \$DBD::Pg::err,
'Errstr' => \$DBD::Pg::errstr,
'State' => \$DBD::Pg::sqlstate,
'Attribution' => "DBD::Pg $realversion by Greg Sabino Mullane and others",
});
# uncoverable branch false
if (!$methods_are_installed) {
DBD::Pg::db->install_method('pg_cancel');
DBD::Pg::db->install_method('pg_endcopy');
DBD::Pg::db->install_method('pg_error_field');
DBD::Pg::db->install_method('pg_getline');
DBD::Pg::db->install_method('pg_getcopydata');
DBD::Pg::db->install_method('pg_getcopydata_async');
DBD::Pg::db->install_method('pg_notifies');
DBD::Pg::db->install_method('pg_putcopydata');
DBD::Pg::db->install_method('pg_putcopyend');
DBD::Pg::db->install_method('pg_ping');
DBD::Pg::db->install_method('pg_putline');
DBD::Pg::db->install_method('pg_ready');
DBD::Pg::db->install_method('pg_release');
DBD::Pg::db->install_method('pg_result'); ## NOT duplicated below!
DBD::Pg::db->install_method('pg_rollback_to');
DBD::Pg::db->install_method('pg_savepoint');
DBD::Pg::db->install_method('pg_server_trace');
DBD::Pg::db->install_method('pg_server_untrace');
DBD::Pg::db->install_method('pg_type_info');
DBD::Pg::st->install_method('pg_cancel');
DBD::Pg::st->install_method('pg_result');
DBD::Pg::st->install_method('pg_ready');
DBD::Pg::st->install_method('pg_canonical_ids');
DBD::Pg::st->install_method('pg_canonical_names');
DBD::Pg::db->install_method('pg_lo_creat');
DBD::Pg::db->install_method('pg_lo_open');
DBD::Pg::db->install_method('pg_lo_write');
DBD::Pg::db->install_method('pg_lo_read');
DBD::Pg::db->install_method('pg_lo_lseek');
DBD::Pg::db->install_method('pg_lo_lseek64');
DBD::Pg::db->install_method('pg_lo_tell');
DBD::Pg::db->install_method('pg_lo_tell64');
DBD::Pg::db->install_method('pg_lo_truncate');
DBD::Pg::db->install_method('pg_lo_truncate64');
DBD::Pg::db->install_method('pg_lo_close');
DBD::Pg::db->install_method('pg_lo_unlink');
DBD::Pg::db->install_method('pg_lo_import');
DBD::Pg::db->install_method('pg_lo_import_with_oid');
DBD::Pg::db->install_method('pg_lo_export');
$methods_are_installed++;
}
return $drh;
} ## end of driver
1;
} ## end of package DBD::Pg
{
package DBD::Pg::dr;
use strict;
## Returns an array of formatted database names from the pg_database table
sub data_sources {
my $drh = shift;
my $conninfo = shift || '';
my $connstring = 'dbname=postgres';
if ($ENV{DBI_DSN}) {
($connstring = $ENV{DBI_DSN}) =~ s/dbi:Pg://i;
}
if (length $conninfo) {
$connstring .= ";$conninfo";
}
my $dbh = DBD::Pg::dr::connect($drh, $connstring) or return;
$dbh->{AutoCommit}=1;
my $SQL = 'SELECT pg_catalog.quote_ident(datname) FROM pg_catalog.pg_database ORDER BY 1';
my $sth = $dbh->prepare($SQL);
$sth->execute();
$conninfo and $conninfo = ";$conninfo";
my @sources = map { "dbi:Pg:dbname=$_->[0]$conninfo" } @{$sth->fetchall_arrayref()};
$dbh->disconnect;
return @sources;
}
sub connect { ## no critic (ProhibitBuiltinHomonyms)
my ($drh, $dbname, $user, $pass, $attr) = @_;
## Allow "db" and "database" as synonyms for "dbname"
$dbname =~ s/\b(?:db|database)\s*=/dbname=/;
my $name = $dbname;
if ($dbname =~ m{dbname\s*=\s*[\"\']([^\"\']+)}) {
$name = "'$1'";
$dbname =~ s/\"/\'/g;
}
elsif ($dbname =~ m{dbname\s*=\s*([^;]+)}) {
$name = $1;
}
$user = defined($user) ? $user : defined $ENV{DBI_USER} ? $ENV{DBI_USER} : '';
$pass = defined($pass) ? $pass : defined $ENV{DBI_PASS} ? $ENV{DBI_PASS} : '';
my ($dbh) = DBI::_new_dbh($drh, {
'Name' => $dbname,
'Username' => $user,
'CURRENT_USER' => $user,
});
# Connect to the database..
DBD::Pg::db::_login($dbh, $dbname, $user, $pass, $attr) or return undef;
my $version = $dbh->{pg_server_version};
$dbh->{private_dbdpg}{version} = $version;
if ($attr) {
if ($attr->{dbd_verbose}) {
$dbh->trace('DBD');
}
}
return $dbh;
}
sub private_attribute_info {
return {
};
}
} ## end of package DBD::Pg::dr
{
package DBD::Pg::db;
use DBI qw(:sql_types);
use strict;
sub parse_trace_flag {
return DBD::Pg->parse_trace_flag($_[1]);
}
sub prepare {
my($dbh, $statement, @attribs) = @_;
return undef if ! defined $statement;
# Create a 'blank' statement handle:
my $sth = DBI::_new_sth($dbh, {
'Statement' => $statement,
});
DBD::Pg::st::_prepare($sth, $statement, @attribs);
return $sth;
}
sub last_insert_id {
my ($dbh, undef, $schema, $table, undef, $attr) = @_;
## Our ultimate goal is to get a sequence
my ($sth, $count, $SQL, $sequence);
## Cache all of our table lookups? Default is yes
my $cache = 1;
## Catalog and col (arguments 2 and 5) are not used
$schema = '' if ! defined $schema;
$table = '' if ! defined $table;
my $cachename = join("\0", 'lii', $schema, $table);
if (defined $attr and length $attr) {
## If not a hash, assume it is a sequence name
if (! ref $attr) {
$attr = {sequence => $attr};
}
elsif (ref $attr ne 'HASH') {
$dbh->set_err(1, 'last_insert_id must be passed a hashref as the final argument');
return undef;
}
## Named sequence overrides any table or schema settings
if (exists $attr->{sequence} and length $attr->{sequence}) {
$sequence = $attr->{sequence};
}
if (exists $attr->{pg_cache}) {
$cache = $attr->{pg_cache};
}
}
if (! defined $sequence and exists $dbh->{private_dbdpg}{$cachename} and $cache) {
$sequence = $dbh->{private_dbdpg}{$cachename};
}
elsif (! defined $sequence) {
## At this point, we must have a valid table name
if (! length $table) {
$dbh->set_err(1, 'last_insert_id needs at least a sequence or table name');
return undef;
}
my @args = ($table);
my $schemawhere;
if (length $schema) {
# if given a schema, use that
$schemawhere = 'n.nspname = ?';
push @args, $schema;
} else {
# otherwise it must be visible via the search path
$schemawhere = 'pg_catalog.pg_table_is_visible(c.oid)';
}
## Is there a sequence associated with the table via a unique, indexed column,
## either via ownership (e.g. serial, identity) or a manual default?
my $idcond = $dbh->{private_dbdpg}{version} >= 100000
? q{a.attidentity <> ''} : q{false};
$SQL = sprintf(q{
SELECT i.indisprimary,
COALESCE(
-- this takes the table name as text, not regclass
pg_catalog.pg_get_serial_sequence(
-- and pre-8.3 doesn't have a cast from regclass to text,
-- and pre-9.3 doesn't have format, so do it the long way
quote_ident(n.nspname) || '.' || quote_ident(c.relname),
a.attname),
(SELECT replace(substring(pg_catalog.pg_get_expr(d.adbin, d.adrelid)
from $r$^nextval\('(.+)'::[\w\s]+\)$$r$),
-- unescape any single quotes from the default
$$''$$, $$'$$)
FROM pg_catalog.pg_attrdef d
WHERE a.atthasdef
AND a.attrelid = d.adrelid
AND a.attnum = d.adnum)
) AS seqname
FROM pg_class c
JOIN pg_catalog.pg_namespace n ON (n.oid = c.relnamespace)
-- LEFT JOIN so we can distingiuish between table not found (zero rows)
-- and no suitable column found (at least one all-NULL row)
LEFT JOIN pg_catalog.pg_index i
ON c.oid = i.indrelid AND i.indisunique
LEFT JOIN pg_catalog.pg_attribute a
ON i.indrelid = a.attrelid AND i.indkey[0]=a.attnum
AND (a.atthasdef OR %s)
WHERE c.relname = ? AND %s
}, $idcond, $schemawhere);
$sth = $dbh->prepare_cached($SQL);
$count = $sth->execute(@args);
if (!defined $count or $count eq '0E0') {
$sth->finish();
my $message = qq{Could not find the table "$table"};
length $schema and $message .= qq{ in the schema "$schema"};
$dbh->set_err(1, $message);
return undef;
}
my $info = $sth->fetchall_arrayref();
## We have at least one with a default value. See if we found any sequences
my @def = grep { defined $_->[1] } @$info;
if (!@def) {
## This may be an inherited table, in which case we can use the parent's info
$SQL = 'SELECT inhparent::regclass FROM pg_inherits WHERE inhrelid = ?::regclass::oid';
my $isth = $dbh->prepare($SQL);
$count = $isth->execute($table);
if ($count < 1) {
$isth->finish();
$dbh->set_err(1, qq{No suitable column found for last_insert_id of table "$table"\n});
return undef;
}
my $parent = $isth->fetch->[0];
$args[0] = $parent;
$count = $sth->execute(@args);
if (1 == $count) {
$info = $sth->fetchall_arrayref();
@def = grep { defined $_->[1] } @$info;
}
if (!@def) {
$sth->finish();
$dbh->set_err(1, qq{No suitable column found for last_insert_id of table "$table"\n});
return undef;
}
## Fall through with inherited information
}
## Tiebreaker goes to the primary keys
if (@def > 1) {
my @pri = grep { $_->[0] } @def;
if (1 != @pri) {
$dbh->set_err(1, qq{No suitable column found for last_insert_id of table "$table"\n});
return undef;
}
@def = @pri;
}
$sequence = $def[0]->[1];
## Cache this information for subsequent calls
$dbh->{private_dbdpg}{$cachename} = $sequence;
}
$sth = $dbh->prepare_cached('SELECT pg_catalog.currval(?)');
$count = $sth->execute($sequence);
return undef if ! defined $count;
return $sth->fetchall_arrayref()->[0][0];
} ## end of last_insert_id
sub ping {
my $dbh = shift;
local $SIG{__WARN__} if $dbh->FETCH('PrintError');
my $ret = DBD::Pg::db::_ping($dbh);
return $ret < 1 ? 0 : $ret;
}
sub pg_ping {
my $dbh = shift;
local $SIG{__WARN__} if $dbh->FETCH('PrintError');
return DBD::Pg::db::_ping($dbh);
}
sub pg_type_info {
my($dbh,$pg_type) = @_;
local $SIG{__WARN__} if $dbh->FETCH('PrintError');
return DBD::Pg::db::_pg_type_info($pg_type);
}
# Column expected in statement handle returned.
# table_cat, table_schem, table_name, column_name, data_type, type_name,
# column_size, buffer_length, DECIMAL_DIGITS, NUM_PREC_RADIX, NULLABLE,
# REMARKS, COLUMN_DEF, SQL_DATA_TYPE, SQL_DATETIME_SUB, CHAR_OCTET_LENGTH,
# ORDINAL_POSITION, IS_NULLABLE
# The result set is ordered by TABLE_SCHEM, TABLE_NAME and ORDINAL_POSITION.
sub column_info {
my $dbh = shift;
my (undef, $schema, $table, $column) = @_;
my @search;
## If the schema or table has an underscore or a %, use a LIKE comparison
if (defined $schema and length $schema) {
push @search, 'n.nspname ' . ($schema =~ /[_%]/ ? 'LIKE ' : '= ') .
$dbh->quote($schema);
}
if (defined $table and length $table) {
push @search, 'c.relname ' . ($table =~ /[_%]/ ? 'LIKE ' : '= ') .
$dbh->quote($table);
}
if (defined $column and length $column) {
push @search, 'a.attname ' . ($column =~ /[_%]/ ? 'LIKE ' : '= ') .
$dbh->quote($column);
}
my $whereclause = join "\n\t\t\t\tAND ", '', @search;
my $col_info_sql = qq!
SELECT
pg_catalog.quote_ident(pg_catalog.current_database()) AS "TABLE_CAT"
, pg_catalog.quote_ident(n.nspname) AS "TABLE_SCHEM"
, pg_catalog.quote_ident(c.relname) AS "TABLE_NAME"
, pg_catalog.quote_ident(a.attname) AS "COLUMN_NAME"
, a.atttypid AS "DATA_TYPE"
, pg_catalog.format_type(a.atttypid, NULL) AS "TYPE_NAME"
, a.attlen AS "COLUMN_SIZE"
, NULL::text AS "BUFFER_LENGTH"
, NULL::text AS "DECIMAL_DIGITS"
, NULL::text AS "NUM_PREC_RADIX"
, CASE a.attnotnull WHEN 't' THEN 0 ELSE 1 END AS "NULLABLE"
, pg_catalog.col_description(a.attrelid, a.attnum) AS "REMARKS"
, pg_catalog.pg_get_expr(af.adbin, af.adrelid) AS "COLUMN_DEF"
, NULL::text AS "SQL_DATA_TYPE"
, NULL::text AS "SQL_DATETIME_SUB"
, NULL::text AS "CHAR_OCTET_LENGTH"
, a.attnum AS "ORDINAL_POSITION"
, CASE a.attnotnull WHEN 't' THEN 'NO' ELSE 'YES' END AS "IS_NULLABLE"
, pg_catalog.format_type(a.atttypid, a.atttypmod) AS "pg_type"
, '?' AS "pg_constraint"
, n.nspname AS "pg_schema"
, c.relname AS "pg_table"
, a.attname AS "pg_column"
, a.attrelid AS "pg_attrelid"
, a.attnum AS "pg_attnum"
, a.atttypmod AS "pg_atttypmod"
, t.typtype AS "_pg_type_typtype"
, t.oid AS "_pg_type_oid"
FROM
pg_catalog.pg_type t
JOIN pg_catalog.pg_attribute a ON (t.oid = a.atttypid)
JOIN pg_catalog.pg_class c ON (a.attrelid = c.oid)
LEFT JOIN pg_catalog.pg_attrdef af ON (a.attnum = af.adnum AND a.attrelid = af.adrelid)
JOIN pg_catalog.pg_namespace n ON (n.oid = c.relnamespace)
WHERE
a.attnum >= 0
AND c.relkind IN ('r','p','v','m','f')
$whereclause
ORDER BY "TABLE_SCHEM", "TABLE_NAME", "ORDINAL_POSITION"
!;
my $data = $dbh->selectall_arrayref($col_info_sql);
# To turn the data back into a statement handle, we need
# to fetch the data as an array of arrays, and also have a
# a matching array of all the column names
my %col_map = (qw/
TABLE_CAT 0
TABLE_SCHEM 1
TABLE_NAME 2
COLUMN_NAME 3
DATA_TYPE 4
TYPE_NAME 5
COLUMN_SIZE 6
BUFFER_LENGTH 7
DECIMAL_DIGITS 8
NUM_PREC_RADIX 9
NULLABLE 10
REMARKS 11
COLUMN_DEF 12
SQL_DATA_TYPE 13
SQL_DATETIME_SUB 14
CHAR_OCTET_LENGTH 15
ORDINAL_POSITION 16
IS_NULLABLE 17
pg_type 18
pg_constraint 19
pg_schema 20
pg_table 21
pg_column 22
pg_enum_values 23
/);
for my $row (@$data) {
my $typoid = pop @$row;
my $typtype = pop @$row;
my $typmod = pop @$row;
my $attnum = pop @$row;
my $aid = pop @$row;
$row->[$col_map{COLUMN_SIZE}] =
_calc_col_size($typmod,$row->[$col_map{COLUMN_SIZE}]);
# Replace the Pg type with the SQL_ type
$row->[$col_map{DATA_TYPE}] = DBD::Pg::db::pg_type_info($dbh,$row->[$col_map{DATA_TYPE}]);
# Add pg_constraint
my $SQL = q{SELECT pg_catalog.pg_get_constraintdef(oid) }.
q{FROM pg_catalog.pg_constraint WHERE contype = 'c' AND }.
qq{conrelid = $aid AND conkey = '{$attnum}'};
my $info = $dbh->selectall_arrayref($SQL);
if (@$info) {
$row->[$col_map{pg_constraint}] = $info->[0][0];
}
else {
$row->[$col_map{pg_constraint}] = undef;
}
if ( $typtype eq 'e' ) {
my $order_column = $dbh->{private_dbdpg}{version} >= 90100
? 'enumsortorder' : 'oid';
$SQL = "SELECT enumlabel FROM pg_catalog.pg_enum WHERE enumtypid = $typoid ORDER BY $order_column";
$row->[$col_map{pg_enum_values}] = $dbh->selectcol_arrayref($SQL);
}
else {
$row->[$col_map{pg_enum_values}] = undef;
}
}
# Since we've processed the data in Perl, we have to jump through a hoop
# To turn it back into a statement handle
#
return _prepare_from_data(
'column_info',
$data,
[ sort { $col_map{$a} <=> $col_map{$b} } keys %col_map],
);
}
sub _prepare_from_data {
my ($statement, $data, $names, %attrinfo) = @_;
my $sponge = DBI->connect('dbi:Sponge:', '', '', { RaiseError => 1 });
my $sth = $sponge->prepare($statement, { rows=>$data, NAME=>$names, %attrinfo });
return $sth;
}
sub statistics_info {
my $dbh = shift;
my (undef, $schema, $table, $unique_only) = @_;
## Catalog is ignored, but table is mandatory
return undef unless defined $table and length $table;
my $schema_where = '';
my @exe_args = ($table);
my $input_schema = (defined $schema and length $schema) ? 1 : 0;
if ($input_schema) {
$schema_where = 'AND n.nspname = ?';
push(@exe_args, $schema);
}
my $stats_sql;
# Table-level stats
if (!$unique_only) {
$stats_sql .= qq{
SELECT
pg_catalog.current_database() AS "TABLE_CAT",
n.nspname AS "TABLE_SCHEM",
d.relname AS "TABLE_NAME",
NULL AS "NON_UNIQUE",
NULL AS "INDEX_QUALIFIER",
NULL AS "INDEX_NAME",
'table' AS "TYPE",
NULL AS "ORDINAL_POSITION",
NULL AS "COLUMN_NAME",
NULL AS "ASC_OR_DESC",
d.reltuples AS "CARDINALITY",
d.relpages AS "PAGES",
NULL AS "FILTER_CONDITION",
NULL AS "pg_expression",
NULL AS "pg_is_key_column",
NULL AS "pg_null_ordering"
FROM pg_catalog.pg_class d
JOIN pg_catalog.pg_namespace n ON n.oid = d.relnamespace
WHERE d.relname = ? $schema_where
UNION ALL
};
push @exe_args, @exe_args;
}
my $is_key_column = $dbh->{private_dbdpg}{version} >= 110000
? 'col.i <= i.indnkeyatts' : 'true';
my ($asc_or_desc, $null_ordering);
if ($dbh->{private_dbdpg}{version} >= 90600) {
$asc_or_desc = q{
CASE WHEN pg_catalog.pg_index_column_has_property(c.oid, col.i, 'asc') THEN 'A'
WHEN pg_catalog.pg_index_column_has_property(c.oid, col.i, 'desc') THEN 'D'
END};
$null_ordering = q{
CASE WHEN pg_catalog.pg_index_column_has_property(c.oid, col.i, 'nulls_first') THEN 'first'
WHEN pg_catalog.pg_index_column_has_property(c.oid, col.i, 'nulls_last') THEN 'last'
END};
}
elsif ($dbh->{private_dbdpg}{version} > 80300) {
$asc_or_desc = q{
CASE WHEN a.amcanorder THEN
CASE WHEN i.indoption[col.i - 1] & 1 = 0 THEN 'A' ELSE 'D' END
END};
$null_ordering = q{
CASE WHEN a.amcanorder THEN
CASE WHEN i.indoption[col.i - 1] & 2 = 0 THEN 'last' ELSE 'first' END
END};
}
else {
$asc_or_desc = q{CASE WHEN a.amorderstrategy <> 0 THEN 'A' END};
$null_ordering = q{CASE WHEN a.amorderstrategy <> 0 THEN 'last' END};
}
# Fetch the index definitions
$stats_sql .= qq{
SELECT
pg_catalog.current_database() AS "TABLE_CAT",
n.nspname AS "TABLE_SCHEM",
d.relname AS "TABLE_NAME",
NOT(i.indisunique) AS "NON_UNIQUE",
NULL AS "INDEX_QUALIFIER",
c.relname AS "INDEX_NAME",
CASE WHEN i.indisclustered THEN 'clustered'
WHEN a.amname = 'btree' THEN 'btree'
WHEN a.amname = 'hash' THEN 'hashed'
ELSE 'other'
END AS "TYPE",
col.i AS "ORDINAL_POSITION",
att.attname AS "COLUMN_NAME",
$asc_or_desc AS "ASC_OR_DESC",
c.reltuples AS "CARDINALITY",
c.relpages AS "PAGES",
pg_catalog.pg_get_expr(i.indpred,i.indrelid)
AS "FILTER_CONDITION",
pg_catalog.pg_get_indexdef(i.indexrelid, col.i, true)
AS "pg_expression",
$is_key_column AS "pg_is_key_column",
$null_ordering AS "pg_null_ordering"
FROM
pg_catalog.pg_index i
JOIN pg_catalog.pg_class c ON c.oid = i.indexrelid
JOIN pg_catalog.pg_class d ON d.oid = i.indrelid
JOIN pg_catalog.pg_am a ON a.oid = c.relam
JOIN pg_catalog.pg_namespace n ON n.oid = d.relnamespace
JOIN pg_catalog.generate_series(1, pg_catalog.current_setting('max_index_keys')::integer) col(i)
ON col.i <= i.indnatts
LEFT JOIN pg_catalog.pg_attribute att
ON att.attrelid = d.oid AND att.attnum = i.indkey[col.i - 1]
WHERE
d.relname = ? $schema_where
AND (i.indisunique OR NOT(?)) -- unique_only
ORDER BY
"NON_UNIQUE", "TYPE", "INDEX_QUALIFIER", "INDEX_NAME", "ORDINAL_POSITION"
};
my $sth = $dbh->prepare($stats_sql);
$sth->execute(@exe_args, 0+!!$unique_only);
return $sth;
}
sub primary_key_info {
my $dbh = shift;
my (undef, $schema, $table, $attr) = @_;
my @cols = (qw(TABLE_CAT TABLE_SCHEM TABLE_NAME
COLUMN_NAME KEY_SEQ PK_NAME DATA_TYPE
pg_tablespace_name pg_tablespace_location
pg_schema pg_table pg_column
)
);
## Catalog is ignored, but table is mandatory
if (! defined $table or ! length $table) {
return _prepare_from_data('primary_key_info', [], \@cols);
}
my $whereclause = 'AND c.relname = ' . $dbh->quote($table);
if (defined $schema and length $schema) {
$whereclause .= "\n\t\t\tAND n.nspname = " . $dbh->quote($schema);
}
my $pri_key_sql = qq{
SELECT
c.oid
, pg_catalog.quote_ident(n.nspname)
, pg_catalog.quote_ident(c.relname)
, pg_catalog.quote_ident(c2.relname)
, i.indkey
, pg_catalog.quote_ident(t.spcname)
, pg_catalog.quote_ident(t.spclocation)
, n.nspname, c.relname, c2.relname
, pg_catalog.quote_ident(pg_catalog.current_database())
FROM
pg_catalog.pg_class c
JOIN pg_catalog.pg_index i ON (i.indrelid = c.oid)
JOIN pg_catalog.pg_class c2 ON (c2.oid = i.indexrelid)
LEFT JOIN pg_catalog.pg_namespace n ON (n.oid = c.relnamespace)
LEFT JOIN pg_catalog.pg_tablespace t ON (t.oid = c.reltablespace)
WHERE
i.indisprimary IS TRUE
$whereclause
};
if ($dbh->{private_dbdpg}{version} >= 90200) {
$pri_key_sql =~ s/t.spclocation/pg_catalog.pg_tablespace_location(t.oid)/;
}
my $sth = $dbh->prepare($pri_key_sql);
$sth->execute();
my $info = $sth->fetchall_arrayref()->[0];
if (! defined $info) {
return _prepare_from_data('primary_key_info', [], \@cols);
}
# Get the attribute information
my $indkey = join ',', split /\s+/, $info->[4];
my $sql = qq{
SELECT a.attnum, pg_catalog.quote_ident(a.attname) AS colname,
pg_catalog.quote_ident(t.typname) AS typename
FROM pg_catalog.pg_attribute a, pg_catalog.pg_type t
WHERE a.attrelid = '$info->[0]'
AND a.atttypid = t.oid
AND attnum IN ($indkey);
};
$sth = $dbh->prepare($sql);
$sth->execute();
my $attribinfo = $sth->fetchall_hashref('attnum');
my $pkinfo = [];
## Normal way: complete "row" per column in the primary key
if (!exists $attr->{'pg_onerow'}) {
my $x=0;
my @key_seq = split/\s+/, $info->[4];
for (@key_seq) {
# TABLE_CAT
$pkinfo->[$x][0] = $info->[10];
# SCHEMA_NAME
$pkinfo->[$x][1] = $info->[1];
# TABLE_NAME
$pkinfo->[$x][2] = $info->[2];
# COLUMN_NAME
$pkinfo->[$x][3] = $attribinfo->{$_}{colname};
# KEY_SEQ
$pkinfo->[$x][4] = $_;
# PK_NAME
$pkinfo->[$x][5] = $info->[3];
# DATA_TYPE
$pkinfo->[$x][6] = $attribinfo->{$_}{typename};
$pkinfo->[$x][7] = $info->[5];
$pkinfo->[$x][8] = $info->[6];
$pkinfo->[$x][9] = $info->[7];
$pkinfo->[$x][10] = $info->[8];
$pkinfo->[$x][11] = $info->[9];
$x++;
}
}
else { ## Nicer way: return only one row
# TABLE_CAT
$info->[0] = $info->[10];
# TABLESPACES
$info->[7] = $info->[5];
$info->[8] = $info->[6];
# Unquoted names
$info->[9] = $info->[7];
$info->[10] = $info->[8];
$info->[11] = $info->[9];
# PK_NAME
$info->[5] = $info->[3];
# COLUMN_NAME
$info->[3] = 2==$attr->{'pg_onerow'} ?
[ map { $attribinfo->{$_}{colname} } split /\s+/, $info->[4] ] :
join ', ', map { $attribinfo->{$_}{colname} } split /\s+/, $info->[4];
# DATA_TYPE
$info->[6] = 2==$attr->{'pg_onerow'} ?
[ map { $attribinfo->{$_}{typename} } split /\s+/, $info->[4] ] :
join ', ', map { $attribinfo->{$_}{typename} } split /\s+/, $info->[4];
# KEY_SEQ
$info->[4] = 2==$attr->{'pg_onerow'} ?
[ split /\s+/, $info->[4] ] :
join ', ', split /\s+/, $info->[4];
$pkinfo = [$info];
}
return _prepare_from_data('primary_key_info', $pkinfo, \@cols);
}
sub primary_key {
my $sth = primary_key_info(@_[0..3], {pg_onerow => 2});
my $result = $sth->fetchall_arrayref();
return defined $result->[0] ? @{$result->[0][3]} : ();
}
sub foreign_key_info {
my $dbh = shift;
## PK: catalog, schema, table, FK: catalog, schema, table, attr
## Each of these may be undef or empty
my $pschema = $_[1] || '';
my $ptable = $_[2] || '';
my $fschema = $_[4] || '';
my $ftable = $_[5] || '';
my @cols = (qw(
UK_TABLE_CAT UK_TABLE_SCHEM UK_TABLE_NAME UK_COLUMN_NAME
FK_TABLE_CAT FK_TABLE_SCHEM FK_TABLE_NAME FK_COLUMN_NAME
ORDINAL_POSITION UPDATE_RULE DELETE_RULE FK_NAME UK_NAME
DEFERABILITY UNIQUE_OR_PRIMARY UK_DATA_TYPE FK_DATA_TYPE
));
if ($dbh->{FetchHashKeyName} eq 'NAME_lc') {
for my $col (@cols) {
$col = lc $col;
}
}
## Must have at least one named table
if (!length($ptable) and !length($ftable)) {
return _prepare_from_data('foreign_key_info', [], \@cols);
}
## If only the primary table is given, we return only those columns
## that are used as foreign keys, even if that means that we return
## unique keys but not primary one. We also return all the foreign
## tables/columns that are referencing them, of course.
## If no schema is given, respect search_path by using pg_table_is_visible()
my @where;
for ([$ptable, $pschema, 'uk'], [$ftable, $fschema, 'fk']) {
my ($table, $schema, $type) = @$_;
if (length $table) {
push @where, "${type}_class.relname = " . $dbh->quote($table);
if (length $schema) {
push @where, "${type}_ns.nspname = " . $dbh->quote($schema);
}
else {
push @where, "pg_catalog.pg_table_is_visible(${type}_class.oid)"
}
}
}
my $WHERE = join ' AND ', @where;
my $SQL = qq{
SELECT
pg_catalog.quote_ident(pg_catalog.current_database()),
pg_catalog.quote_ident(uk_ns.nspname),
pg_catalog.quote_ident(uk_class.relname),
pg_catalog.quote_ident(uk_col.attname),
pg_catalog.quote_ident(pg_catalog.current_database()),
pg_catalog.quote_ident(fk_ns.nspname),
pg_catalog.quote_ident(fk_class.relname),
pg_catalog.quote_ident(fk_col.attname),
colnum.i,
CASE constr.confupdtype
WHEN 'c' THEN 0 WHEN 'r' THEN 1 WHEN 'n' THEN 2 WHEN 'a' THEN 3 WHEN 'd' THEN 4 ELSE -1
END,
CASE constr.confdeltype
WHEN 'c' THEN 0 WHEN 'r' THEN 1 WHEN 'n' THEN 2 WHEN 'a' THEN 3 WHEN 'd' THEN 4 ELSE -1
END,
pg_catalog.quote_ident(constr.conname), pg_catalog.quote_ident(uk_constr.conname),
CASE
WHEN constr.condeferrable = 'f' THEN 7
WHEN constr.condeferred = 't' THEN 6
WHEN constr.condeferred = 'f' THEN 5
ELSE -1
END,
CASE coalesce(uk_constr.contype, 'u')
WHEN 'u' THEN 'UNIQUE' WHEN 'p' THEN 'PRIMARY'
END,
pg_catalog.quote_ident(uk_type.typname), pg_catalog.quote_ident(fk_type.typname)
FROM pg_catalog.pg_constraint constr
JOIN pg_catalog.pg_class uk_class ON constr.confrelid = uk_class.oid
JOIN pg_catalog.pg_namespace uk_ns ON uk_class.relnamespace = uk_ns.oid
JOIN pg_catalog.pg_class fk_class ON constr.conrelid = fk_class.oid
JOIN pg_catalog.pg_namespace fk_ns ON fk_class.relnamespace = fk_ns.oid
-- can't do unnest() until 8.4, and would need WITH ORDINALITY to get the array indices,
-- wich isn't available until 9.4 at the earliest, so we join against a series table instead
JOIN pg_catalog.generate_series(1, pg_catalog.current_setting('max_index_keys')::integer) colnum(i)
ON colnum.i <= pg_catalog.array_upper(constr.conkey,1)
JOIN pg_catalog.pg_attribute uk_col ON uk_col.attrelid = constr.confrelid AND uk_col.attnum = constr.confkey[colnum.i]
JOIN pg_catalog.pg_type uk_type ON uk_col.atttypid = uk_type.oid
JOIN pg_catalog.pg_attribute fk_col ON fk_col.attrelid = constr.conrelid AND fk_col.attnum = constr.conkey[colnum.i]
JOIN pg_catalog.pg_type fk_type ON fk_col.atttypid = fk_type.oid
-- We can't match confkey from the fk constraint to conkey of the unique constraint,
-- because the unique constraint might not exist or there might be more than one
-- matching one. However, there must be at least a unique _index_ on the key
-- columns, so we look for that; but we can't find it via pg_index, since there may
-- again be more than one matching index.
-- So instead, we look at pg_depend for the dependency that was created by the fk
-- constraint. This dependency is of type 'n' (normal) and ties the pg_constraint
-- row oid to the pg_class oid for the index relation (a single arbitrary one if
-- more than one matching unique index existed at the time the constraint was
-- created). Fortunately, the constraint does not create dependencies on the
-- referenced table itself, but on the _columns_ of the referenced table, so the
-- index can be distinguished easily. Then we look for another pg_depend entry,
-- this time an 'i' (implementation) dependency from a pg_constraint oid (the unique
-- constraint if one exists) to the index oid; but we have to allow for the
-- possibility that this one doesn't exist. - Andrew Gierth (RhodiumToad)
JOIN pg_catalog.pg_depend dep ON (