Allow timescale to be a non-primary rails db

Author: segiddins

lib/timescaledb/rails/extensions/active_record/schema_dumper.rb

@@ -2,6 +2,7 @@
 
 require 'active_record/connection_adapters/postgresql_adapter'
 require 'timescaledb/rails/orderby_compression'
+require 'tsort'
 
 module Timescaledb
   module Rails
@@ -19,14 +20,20 @@ def tables(stream)
         def continuous_aggregates(stream)
           return unless timescale_enabled?
 
-          Timescaledb::Rails::ContinuousAggregate.all.each do |continuous_aggregate|
-            continuous_aggregate(continuous_aggregate, stream)
-            continuous_aggregate_policy(continuous_aggregate, stream)
+          deps = Timescaledb::Rails::ContinuousAggregate.find_each.index_by(&:materialization_hypertable_name)
+
+          TSort.tsort(
+            ->(&b) { deps.each_value(&b) },
+            ->(n, &b) { Array.wrap(deps[n.hypertable_name]).each(&b) }
+          ).each do |ca|
+            continuous_aggregate(ca, stream)
+            continuous_aggregate_policy(ca, stream)
           end
         end
 
-        def continuous_aggregate(continuous_aggregate, stream)
-          stream.puts "  create_continuous_aggregate #{continuous_aggregate.view_name.inspect}, <<-SQL"
+        def continuous_aggregate(continuous_aggregate, stream, force: false)
+          stream.puts "  create_continuous_aggregate #{continuous_aggregate.view_name.inspect}, <<-SQL, " \
+                      "force: #{force.inspect}"
           stream.puts "  #{continuous_aggregate.view_definition.strip.indent(2)}"
           stream.puts '  SQL'
           stream.puts
@@ -149,7 +156,8 @@ def format_hypertable_option_value(value)
         end
 
         def timescale_enabled?
-          Timescaledb::Rails::Hypertable.table_exists?
+          @connection.pool.db_config.name == Timescaledb::Rails::Hypertable.connection.pool.db_config.name &&
+            Timescaledb::Rails::Hypertable.table_exists?
         end
       end
     end

lib/timescaledb/rails/extensions/active_record/schema_statements.rb

@@ -116,8 +116,15 @@ def remove_hypertable_reorder_policy(table_name, _index_name = nil)
         #     'temperature_events', "SELECT * FROM events where event_type = 'temperature'"
         #   )
         #
-        def create_continuous_aggregate(view_name, view_query)
-          execute "CREATE MATERIALIZED VIEW #{view_name} WITH (timescaledb.continuous) AS #{view_query};"
+        def create_continuous_aggregate(view_name, view_query, force: false)
+          if force
+            execute "DROP MATERIALIZED VIEW #{quote_table_name(view_name)} CASCADE;" if view_exists? view_name
+          else
+            schema_cache.clear_data_source_cache!(view_name.to_s)
+          end
+
+          execute "CREATE MATERIALIZED VIEW #{quote_table_name(view_name)} " \
+                  "WITH (timescaledb.continuous) AS #{view_query};"
         end
 
         # Drops a continuous aggregate

lib/timescaledb/rails/model/hyperfunctions.rb

@@ -10,12 +10,18 @@ module Hyperfunctions
         TIME_BUCKET_ALIAS = 'time_bucket'
 
         # @return [ActiveRecord::Relation<ActiveRecord::Base>]
-        def time_bucket(interval, target_column = nil)
-          target_column ||= hypertable_time_column_name
+        def time_bucket(interval, target_column = nil, select_alias: TIME_BUCKET_ALIAS)
+          target_column &&= Arel.sql(target_column.to_s)
+          target_column ||= arel_table[hypertable_time_column_name]
 
-          select("time_bucket('#{format_interval_value(interval)}', #{target_column}) as #{TIME_BUCKET_ALIAS}")
-            .group(TIME_BUCKET_ALIAS)
-            .order(TIME_BUCKET_ALIAS)
+          time_bucket = Arel::Nodes::NamedFunction.new(
+            'time_bucket',
+            [Arel::Nodes.build_quoted(format_interval_value(interval)), target_column]
+          )
+
+          select(time_bucket.dup.as(select_alias))
+            .group(time_bucket)
+            .order(time_bucket)
             .extending(AggregateFunctions)
         end
 

lib/timescaledb/rails/models/chunk.rb

@@ -3,7 +3,7 @@
 module Timescaledb
   module Rails
     # :nodoc:
-    class Chunk < ::ActiveRecord::Base
+    class Chunk < Railtie.config.record_base.constantize
       self.table_name = 'timescaledb_information.chunks'
       self.primary_key = 'hypertable_name'
 
@@ -17,13 +17,13 @@ def chunk_full_name
       end
 
       def compress!
-        ::ActiveRecord::Base.connection.execute(
+        self.class.connection.execute(
           "SELECT compress_chunk('#{chunk_full_name}')"
         )
       end
 
       def decompress!
-        ::ActiveRecord::Base.connection.execute(
+        self.class.connection.execute(
           "SELECT decompress_chunk('#{chunk_full_name}')"
         )
       end
@@ -40,7 +40,7 @@ def reorder!(index = nil)
         options = ["'#{chunk_full_name}'"]
         options << "'#{index}'" if index.present?
 
-        ::ActiveRecord::Base.connection.execute(
+        self.class.connection.execute(
           "SELECT reorder_chunk(#{options.join(', ')})"
         )
       end

lib/timescaledb/rails/models/compression_setting.rb

@@ -3,7 +3,7 @@
 module Timescaledb
   module Rails
     # :nodoc:
-    class CompressionSetting < ::ActiveRecord::Base
+    class CompressionSetting < Railtie.config.record_base.constantize
       self.table_name = 'timescaledb_information.compression_settings'
       self.primary_key = 'hypertable_name'
 

lib/timescaledb/rails/models/concerns/durationable.rb

@@ -11,6 +11,8 @@ module Durationable
 
         # @return [String]
         def parse_duration(duration)
+          return if duration.nil?
+
           duration_in_seconds = duration_in_seconds(duration)
 
           duration_to_interval(

lib/timescaledb/rails/models/continuous_aggregate.rb

@@ -5,7 +5,7 @@
 module Timescaledb
   module Rails
     # :nodoc:
-    class ContinuousAggregate < ::ActiveRecord::Base
+    class ContinuousAggregate < Railtie.config.record_base.constantize
       include Timescaledb::Rails::Models::Durationable
 
       self.table_name = 'timescaledb_information.continuous_aggregates'
@@ -19,7 +19,7 @@ class ContinuousAggregate < ::ActiveRecord::Base
       # @param [DateTime] end_time
       #
       def refresh!(start_time = 'NULL', end_time = 'NULL')
-        ::ActiveRecord::Base.connection.execute(
+        self.class.connection.execute(
           "CALL refresh_continuous_aggregate('#{view_name}', #{start_time}, #{end_time});"
         )
       end

lib/timescaledb/rails/models/dimension.rb

@@ -3,7 +3,7 @@
 module Timescaledb
   module Rails
     # :nodoc:
-    class Dimension < ::ActiveRecord::Base
+    class Dimension < Railtie.config.record_base.constantize
       TIME_TYPE = 'Time'
 
       self.table_name = 'timescaledb_information.dimensions'

lib/timescaledb/rails/models/hypertable.rb

@@ -5,7 +5,7 @@
 module Timescaledb
   module Rails
     # :nodoc:
-    class Hypertable < ::ActiveRecord::Base
+    class Hypertable < Railtie.config.record_base.constantize
       include Timescaledb::Rails::Models::Durationable
 
       self.table_name = 'timescaledb_information.hypertables'

lib/timescaledb/rails/models/job.rb

@@ -3,7 +3,7 @@
 module Timescaledb
   module Rails
     # :nodoc:
-    class Job < ::ActiveRecord::Base
+    class Job < Railtie.config.record_base.constantize
       self.table_name = 'timescaledb_information.jobs'
       self.primary_key = 'hypertable_name'