Pg: Add $dbh.column-info for schema introspection

lib/DBDish/Pg.pm6

@@ -1,7 +1,7 @@
 use v6;
 need DBDish;
 
-unit class DBDish::Pg:auth<mberends>:ver<0.1.5> does DBDish::Driver;
+unit class DBDish::Pg:auth<mberends>:ver<0.1.6> does DBDish::Driver;
 use DBDish::Pg::Native;
 need DBDish::Pg::Connection;
 

lib/DBDish/Pg/Connection.pm6

@@ -4,6 +4,8 @@ need DBDish;
 unit class DBDish::Pg::Connection does DBDish::Connection;
 use DBDish::Pg::Native;
 need DBDish::Pg::StatementHandle;
+need DBDish::TestMock;
+constant SQLType = DBIish::SQLType;
 
 has PGconn $!pg_conn is required handles <
     pg-notifies pg-socket pg-parameter-status
@@ -18,6 +20,7 @@ method prepare(Str $statement, *%args) {
     state $statement_postfix = 0;
     my $statement_name = join '_', 'pg', $*PID, $statement_postfix++;
     my $munged = DBDish::Pg::pg-replace-placeholder($statement);
+    die "Can't prepare this: '$statement'!" unless $munged;
     my $result = $!pg_conn.PQprepare($statement_name, $munged, 0, OidArray);
     LEAVE { $result.PQclear if $result }
     if $result && $result.is-ok {
@@ -127,6 +130,156 @@ method _disconnect() {
     $!pg_conn = Nil;
 }
 
+constant %pg-to-sql is export = Map.new: map(
+    { +PGTypes::{.key} => .value }, (
+  PG_BOOL        => SQLType::SQL_BOOLEAN,
+  PG_BPCHAR      => SQLType::SQL_CHAR,
+  PG_BYTEA       => SQLType::SQL_VARBINARY,
+  PG_CHAR        => SQLType::SQL_CHAR,
+  PG_DATE        => SQLType::SQL_TYPE_DATE,
+  PG_FLOAT8      => SQLType::SQL_FLOAT,
+  PG_INT2        => SQLType::SQL_SMALLINT,
+  PG_INT4        => SQLType::SQL_INTEGER,
+  PG_INT8        => SQLType::SQL_BIGINT,
+  PG_NAME        => SQLType::SQL_VARCHAR,
+  PG_NUMERIC     => SQLType::SQL_DECIMAL,
+  PG_TEXT        => SQLType::SQL_LONGVARCHAR,
+  PG_TIME        => SQLType::SQL_TYPE_TIME,
+  PG_TIMESTAMP   => SQLType::SQL_TIMESTAMP,
+  PG_TIMESTAMPTZ => SQLType::SQL_TYPE_TIMESTAMP_WITH_TIMEZONE,
+  PG_VARCHAR     => SQLType::SQL_VARCHAR,
+));
+
+my sub calc-col-size($mod, $size) {
+    if $size.defined && $size > 0 {
+        $size;
+    } elsif $mod > 0xffff {
+        my $prec = ($mod +& 0xffff) - 4;
+        $mod +>= 16;
+        #my $dig = $mod;
+        "$prec,$mod";
+    } elsif $mod >= 4 {
+        $mod - 4;
+    } else {
+        $size
+    }
+}
+
+my sub prepare-from-data($parent, $statement, List() $rows, $col-names, $col-types) {
+    my $mock = DBDish::TestMock.new(:$parent).connect
+        .prepare('col-info',:$rows,:$col-names,:$col-types);
+    $mock.execute;
+    $mock;
+}
+
+# If the ID has an underscore or a %, use a LIKE comparison
+method !make-comp($id, $field) {
+    "$field " ~ ($id ~~ / '_' | '%' / ?? 'LIKE ' !! '= ') ~ self.quote($id);
+}
+
+method column-info(:$catalog, :$schema, :$table, :$column) {
+    my @search = '';
+    @search.push(self!make-comp($schema, 'n.nspname')) if $schema;
+    @search.push(self!make-comp($table,  'c.relname')) if $table;
+    @search.push(self!make-comp($column, 'a.attname')) if $column;
+
+    my $column_def = self.server-version ge v8.0.0
+            ?? 'pg_catalog.pg_get_expr(af.adbin, af.adrelid)'
+            !! 'af.adsrc';
+
+    my $col-info-sql = qq«
+      SELECT
+        NULL::text AS "TABLE_CAT"
+        , quote_ident(n.nspname) AS "TABLE_SCHEM"
+        , quote_ident(c.relname) AS "TABLE_NAME"
+        , 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"
+        , $column_def 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_typtype"
+        , t.oid AS "_pg_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','v','m'){ @search.join("\n\tAND ") }
+      ORDER BY "TABLE_SCHEM", "TABLE_NAME", "ORDINAL_POSITION"»;
+
+    my $sth = self.prepare($col-info-sql);
+    my %col-map := ($sth.column-names Z=> (0..*)).Map;
+
+    my $sth-info = self.prepare(q{
+        SELECT consrc FROM pg_catalog.pg_constraint
+        WHERE contype = 'c' AND conrelid = ? AND conkey = ?
+    });
+
+    $sth.execute;
+    # We need to process the data
+    my $data = $sth.allrows.map(-> @row {
+        # The last five are internal only
+        my ($aid, $attnum, $typmod, $typtype, $typoid) =
+          @row[%col-map<pg_attrelid pg_attnum pg_atttypmod _pg_typtype _pg_oid>]:delete;
+
+        @row[%col-map<COLUMN_SIZE>] = calc-col-size($typmod, @row[%col-map<COLUMN_SIZE>]);
+
+        # Replace the Pg type oid with the PG_/SQL_ type
+        with PGTypes(@row[%col-map<DATA_TYPE>]) {
+            @row[%col-map<DATA_TYPE>] = $_;
+            @row[%col-map<SQL_DATA_TYPE>] = %pg-to-sql{+$_} || SQLType::SQL_UNKNOWN_TYPE;
+        }
+
+        # Add pg_constraint
+        with $sth-info.execute($aid, $attnum) && $sth-info.allrows {
+            @row[%col-map<pg_constraint>] = $_ ?? $_[0][0] !! Any;
+        }
+
+        if $typtype eq 'e'  {
+            @row.push: self.selectcol_arrayref( #TODO remove legacy
+                "SELECT enumlabel FROM pg_catalog.pg_enum WHERE enumtypid = $typoid ORDER BY " ~
+                (self.server-version ge v9.1.0 ?? 'enumsortorder' !! 'oid')
+            );
+        }
+        else {
+            @row.push: Any;
+        }
+        @row;
+    });
+
+    # Since we've processed the data in Perl, we have to jump through a hoop
+    # to turn it back into a statement handle
+    my @col-types = (|$sth.column-types[^23], Str);
+    @col-types[%col-map<DATA_TYPE>, %col-map<SQL_DATA_TYPE>] = Mu;
+    prepare-from-data(self.drv.parent,
+         'column_info',
+         $data,
+         $(|$sth.column-names[^23], 'pg_enum_values'),
+         @col-types
+    );
+}
+
 method table-info(:$catalog, :$schema, :$table, :$type) {
 
     my $tbl_sql;
@@ -231,19 +384,18 @@ method table-info(:$catalog, :$schema, :$table, :$type) {
               !! 't.spclocation'
             ) ~ ') AS "pg_tablespace_location"';
 
-        ## If the schema or table has an underscore or a %, use a LIKE comparison
-        if ($schema.defined && $schema.chars) {
-            @search.push: 'n.nspname ' ~ ($schema ~~ / '_' | '%' / ?? 'LIKE ' !! '= ') ~ self.quote($schema);
+        if $schema.defined && $schema.chars {
+            @search.push: self!make-comp($schema, 'n.nspname');
         }
-        if ($table.defined && $table.chars) {
-            @search.push: 'c.relname ' ~ ($table ~~ / '_' | '%' / ?? 'LIKE ' !! '= ') ~ self.quote($table);
+        if $table.defined && $table.chars {
+            @search.push: self!make-comp($table, 'c.relname');
         }
 
         my $TSJOIN = self.server-version lt v8.0.0
             ?? '(SELECT 0 AS oid, 0 AS spcname, 0 AS spclocation LIMIT 0) AS t ON (t.oid=1)'
             !! 'pg_catalog.pg_tablespace t ON (t.oid = c.reltablespace)';
 
-        my $whereclause = @search.join("\n\t\t\t\t\t AND ");
+        my $whereclause = @search.join("\n\t\t\t\t\tAND ");
         $tbl_sql = qq{
             SELECT NULL::text AS "TABLE_CAT"
              , quote_ident(n.nspname) AS "TABLE_SCHEM"