indexer.rb in pgdexter-0.4.3

- old
+ new

@@ -12,10 +12,11 @@
       @log_explain = options[:log_explain]
       @min_time = options[:min_time] || 0
       @min_calls = options[:min_calls] || 0
       @analyze = options[:analyze]
       @min_cost_savings_pct = options[:min_cost_savings_pct].to_i
+      @log_table = options[:log_table]
       @options = options
       @mutex = Mutex.new
 
       create_extension unless extension_exists?
       execute("SET lock_timeout = '5s'")
@@ -26,11 +27,11 @@
       log "Processing #{queries.size} new query fingerprints"
       process_queries(queries)
     end
 
     def stat_activity
-      execute <<-SQL
+      execute <<~SQL
         SELECT
           pid || ':' || COALESCE(query_start, xact_start) AS id,
           query,
           EXTRACT(EPOCH FROM NOW() - COALESCE(query_start, xact_start)) * 1000.0 AS duration_ms
         FROM
@@ -42,10 +43,40 @@
         ORDER BY
           1
       SQL
     end
 
+    # works with
+    # file_fdw: https://www.postgresql.org/docs/current/file-fdw.html
+    # log_fdw: https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Appendix.PostgreSQL.CommonDBATasks.Extensions.foreign-data-wrappers.html
+    def csvlog_activity(last_log_time)
+      query = <<~SQL
+        SELECT
+          log_time,
+          message,
+          detail
+        FROM
+          #{conn.quote_ident(@log_table)}
+        WHERE
+          log_time >= \$1
+      SQL
+      execute(query, params: [last_log_time])
+    end
+
+    # works with
+    # file_fdw: https://www.postgresql.org/docs/current/file-fdw.html
+    # log_fdw: https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Appendix.PostgreSQL.CommonDBATasks.Extensions.foreign-data-wrappers.html
+    def stderr_activity
+      query = <<~SQL
+        SELECT
+          log_entry
+        FROM
+          #{conn.quote_ident(@log_table)}
+      SQL
+      execute(query)
+    end
+
     def process_queries(queries)
       # reset hypothetical indexes
       reset_hypothetical_indexes
 
       tables = Set.new(database_tables + materialized_views)
@@ -139,11 +170,11 @@
     end
 
     def analyze_tables(tables)
       tables = tables.to_a.sort
 
-      analyze_stats = execute <<-SQL
+      analyze_stats = execute <<~SQL
         SELECT
           schemaname || '.' || relname AS table,
           last_analyze,
           last_autoanalyze
         FROM
@@ -527,11 +558,11 @@
       end
     rescue PG::ConnectionBad => e
       raise Dexter::Abort, e.message
     end
 
-    def execute(query, pretty: true)
+    def execute(query, pretty: true, params: [])
       # use exec_params instead of exec for security
       #
       # Unlike PQexec, PQexecParams allows at most one SQL command in the given string.
       # (There can be semicolons in it, but not more than one nonempty command.)
       # This is a limitation of the underlying protocol, but has some usefulness
@@ -539,11 +570,11 @@
       # https://www.postgresql.org/docs/current/static/libpq-exec.html
       query = squish(query) if pretty
       log colorize("[sql] #{query}", :cyan) if @log_sql
 
       @mutex.synchronize do
-        conn.exec_params(query, []).to_a
+        conn.exec_params(query, params).to_a
       end
     end
 
     def plan(query)
       # strip semi-colons as another measure of defense
@@ -563,11 +594,11 @@
     def create_hypothetical_index(table, col_set)
       execute("SELECT * FROM hypopg_create_index('CREATE INDEX ON #{quote_ident(table)} (#{col_set.map { |c| quote_ident(c[:column])  }.join(", ")})')").first["indexname"]
     end
 
     def database_tables
-      result = execute <<-SQL
+      result = execute <<~SQL
         SELECT
           table_schema || '.' || table_name AS table_name
         FROM
           information_schema.tables
         WHERE
@@ -576,11 +607,11 @@
       result.map { |r| r["table_name"] }
     end
 
     def materialized_views
       if server_version_num >= 90300
-        result = execute <<-SQL
+        result = execute <<~SQL
           SELECT
             schemaname || '.' || matviewname AS table_name
           FROM
             pg_matviews
         SQL
@@ -593,11 +624,11 @@
     def server_version_num
       execute("SHOW server_version_num").first["server_version_num"].to_i
     end
 
     def database_view_tables
-      result = execute <<-SQL
+      result = execute <<~SQL
         SELECT
           schemaname || '.' || viewname AS table_name,
           definition
         FROM
           pg_views
@@ -619,11 +650,11 @@
       view_tables
     end
 
     def stat_statements
       total_time = server_version_num >= 130000 ? "(total_plan_time + total_exec_time)" : "total_time"
-      result = execute <<-SQL
+      result = execute <<~SQL
         SELECT
           DISTINCT query
         FROM
           pg_stat_statements
         INNER JOIN
@@ -665,11 +696,11 @@
     def index_exists?(index)
       indexes([index[:table]]).find { |i| i["columns"] == index[:columns] }
     end
 
     def columns(tables)
-      columns = execute <<-SQL
+      columns = execute <<~SQL
         SELECT
           s.nspname || '.' || t.relname AS table_name,
           a.attname AS column_name,
           pg_catalog.format_type(a.atttypid, a.atttypmod) AS data_type
         FROM pg_attribute a
@@ -684,11 +715,11 @@
 
       columns.map { |v| {table: v["table_name"], column: v["column_name"], type: v["data_type"]} }
     end
 
     def indexes(tables)
-      execute(<<-SQL
+      query = <<~SQL
         SELECT
           schemaname || '.' || t.relname AS table,
           ix.relname AS name,
           regexp_replace(pg_get_indexdef(i.indexrelid), '^[^\\(]*\\((.*)\\)$', '\\1') AS columns,
           regexp_replace(pg_get_indexdef(i.indexrelid), '.* USING ([^ ]*) \\(.*', '\\1') AS using
@@ -706,10 +737,10 @@
           indexprs IS NULL AND
           indpred IS NULL
         ORDER BY
           1, 2
       SQL
-      ).map { |v| v["columns"] = v["columns"].sub(") WHERE (", " WHERE ").split(", ").map { |c| unquote(c) }; v }
+      execute(query).map { |v| v["columns"] = v["columns"].sub(") WHERE (", " WHERE ").split(", ").map { |c| unquote(c) }; v }
     end
 
     def search_path
       execute("SELECT current_schemas(true)")[0]["current_schemas"][1..-2].split(",")
     end