Reversible create_continuous_aggregate

Author: intermittentnrg

lib/timescaledb.rb

@@ -18,6 +18,7 @@
 require_relative 'timescaledb/stats'
 require_relative 'timescaledb/stats_report'
 require_relative 'timescaledb/migration_helpers'
+require_relative 'timescaledb/command_recorder'
 require_relative 'timescaledb/extension'
 require_relative 'timescaledb/version'
 

lib/timescaledb/command_recorder.rb

@@ -0,0 +1,14 @@
+# Useful methods to run TimescaleDB in you Ruby app.
+module Timescaledb
+  # Migration helpers can help you to setup hypertables by default.
+  module CommandRecorder
+    def create_continuous_aggregates(*args)
+      record(:create_continuous_aggregates, args)
+    end
+
+    def invert_create_continuous_aggregates(args)
+      [:drop_continuous_aggregates, args.first]
+    end
+  end
+end
+ActiveRecord::Migration::CommandRecorder.include Timescaledb::CommandRecorder

lib/timescaledb/migration_helpers.rb

@@ -114,16 +114,18 @@ def create_continuous_aggregate(table_name, query, **options)
       create_continuous_aggregate_policy(table_name, **(options[:refresh_policies] || {}))
     end
 
+    alias_method :create_continuous_aggregates, :create_continuous_aggregate
+
     #  Drop a new continuous aggregate.
     #
     #  It basically DROP MATERIALIZED VIEW for a given @name.
     #
     # @param name [String, Symbol] The name of the continuous aggregate view.
-    def drop_continuous_aggregates view_name
+    def drop_continuous_aggregate view_name
       execute "DROP MATERIALIZED VIEW #{view_name}"
     end
 
-    alias_method :create_continuous_aggregates, :create_continuous_aggregate
+    alias_method :drop_continuous_aggregates, :drop_continuous_aggregate
 
     def create_continuous_aggregate_policy(table_name, **options)
       return if options.empty?

spec/timescaledb/migration_helper_spec.rb

@@ -1,4 +1,52 @@
 RSpec.describe Timescaledb::MigrationHelpers, database_cleaner_strategy: :truncation do
+  context 'reversible' do
+    let(:con) { ActiveRecord::Base.connection }
+
+    before(:each) do
+      con.drop_table :ticks, if_exists: true, force: :cascade
+      con.create_table :ticks, hypertable: hypertable_options, id: false do |t|
+        t.string :symbol
+        t.decimal :price
+        t.integer :volume
+        t.timestamps
+      end
+    end
+
+    let(:hypertable_options) do
+      {
+        time_column: 'created_at',
+        chunk_time_interval: '1 min',
+        compress_segmentby: 'symbol',
+        compress_orderby: 'created_at',
+        compress_after: '7 days'
+      }
+    end
+
+    subject(:migration) do
+      Class.new(ActiveRecord::Migration::Current) do
+        def change
+          query = <<~SQL
+            SELECT time_bucket('1m', created_at) as time,
+              symbol,
+              FIRST(price, created_at) as open,
+              MAX(price) as high,
+              MIN(price) as low,
+              LAST(price, created_at) as close,
+              SUM(volume) as volume FROM "ticks" GROUP BY 1,2
+          SQL
+          create_continuous_aggregates('ohlc_1m', query, with_data: true)
+        end
+      end.new
+    end
+
+    it do
+      expect(con).to receive(:create_continuous_aggregates).once.and_call_original
+      expect(con).to receive(:drop_continuous_aggregates).once.and_call_original
+      migration.migrate(:up)
+      migration.migrate(:down)
+    end
+  end
+
   describe ".create_table" do
     let(:con) { ActiveRecord::Base.connection }