feat: add cursor based incremental refresh

server/app/contracts/sync_contracts.rb

@@ -24,6 +24,7 @@ class Create < Dry::Validation::Contract
         required(:sync_interval_unit).filled(:string)
         required(:sync_mode).filled(:string)
         required(:stream_name).filled(:string)
+        optional(:cursor_field).maybe(:string)
 
         # update filled with validating array of hashes
         required(:configuration).filled

server/app/controllers/api/v1/syncs_controller.rb

@@ -93,6 +93,7 @@ def sync_params
                                       :sync_mode,
                                       :sync_interval_unit,
                                       :stream_name,
+                                      :cursor_field,
                                       configuration: %i[from
                                                         to
                                                         mapping_type
@@ -104,7 +105,7 @@ def sync_params
         if params.to_unsafe_h[:sync][:configuration].is_a?(Hash)
           strong_params.merge!(configuration: params.to_unsafe_h[:sync][:configuration])
         end
-
+        strong_params.delete(:cursor_field) if action_name == "update"
         strong_params
       end
     end

server/app/interactors/syncs/create_sync.rb

@@ -5,6 +5,9 @@ class CreateSync
     include Interactor
 
     def call
+      connector = context.workspace.connectors.find_by(id: context.sync_params[:source_id])
+      default_cursor_field = connector&.default_cursor_field(context.sync_params[:stream_name])
+      context.sync_params[:cursor_field] = default_cursor_field if default_cursor_field.present?
       sync = context
              .workspace.syncs
              .create(context.sync_params)

server/app/models/connector.rb

@@ -86,4 +86,9 @@ def connector_query_type
 
     connector_spec&.connector_query_type || "raw_sql"
   end
+
+  def default_cursor_field(stream_name)
+    stream = catalog.catalog["streams"].find { |s| s["name"] == stream_name }
+    stream["default_cursor_field"] if stream && catalog["source_defined_cursor"]
+  end
 end

server/app/models/sync.rb

@@ -13,6 +13,8 @@
 #  source_catalog_id :integer
 #  schedule_type     :string
 #  status            :integer
+#  cursor_field      :string
+#  current_cursor_field :string
 #  created_at        :datetime         not null
 #  updated_at        :datetime         not null
 #
@@ -82,7 +84,9 @@ def to_protocol
         catalog.find_stream_by_name(stream_name)
       ),
       sync_mode: Multiwoven::Integrations::Protocol::SyncMode[sync_mode],
-      destination_sync_mode: Multiwoven::Integrations::Protocol::DestinationSyncMode["insert"]
+      destination_sync_mode: Multiwoven::Integrations::Protocol::DestinationSyncMode["insert"],
+      cursor_field:,
+      current_cursor_field:
     )
   end
 

server/app/serializers/sync_serializer.rb

@@ -3,7 +3,7 @@
 class SyncSerializer < ActiveModel::Serializer
   attributes :id, :source_id, :destination_id, :model_id, :configuration,
              :schedule_type, :sync_mode, :sync_interval, :sync_interval_unit,
-             :stream_name, :status,
+             :stream_name, :status, :cursor_field, :current_cursor_field,
              :updated_at, :created_at
 
   attribute :source do

server/db/migrate/20240412183836_add_cursor_fields_to_syncs.rb

@@ -0,0 +1,6 @@
+class AddCursorFieldsToSyncs < ActiveRecord::Migration[7.1]
+  def change
+    add_column :syncs, :cursor_field, :string
+    add_column :syncs, :current_cursor_field, :string
+  end
+end

server/lib/reverse_etl/extractors/incremental_delta.rb

@@ -17,11 +17,14 @@ def read(sync_run_id, activity)
         batch_query_params = batch_params(source_client, sync_run)
         model = sync_run.sync.model
 
-        ReverseEtl::Utils::BatchQuery.execute_in_batches(batch_query_params) do |records, current_offset|
+        ReverseEtl::Utils::BatchQuery.execute_in_batches(batch_query_params) do |records,
+          current_offset, last_cursor_field_value|
+
           total_query_rows += records.count
           process_records(records, sync_run, model)
           heartbeat(activity)
           sync_run.update(current_offset:, total_query_rows:)
+          sync_run.sync.update(current_cursor_field: last_cursor_field_value)
         end
         # change state querying to queued
         sync_run.queue!

server/lib/reverse_etl/utils/batch_query.rb

@@ -13,15 +13,24 @@ def self.execute_in_batches(params)
           params[:sync_config].limit = params[:batch_size]
           params[:sync_config].offset = current_offset
 
+          CursorQueryBuilder.update_query(params[:sync_config])
+
           # Execute the batch query
           result = params[:client].read(params[:sync_config])
 
+          # Extract the value of the cursor_field column from the last record
+          last_cursor_field_value = nil
+          if params[:sync_config].cursor_field && !result.empty?
+            last_record = result.last.message.record.data
+            last_cursor_field_value = last_record[params[:sync_config].cursor_field]
+          end
+
           # Increment the offset by the batch size for the next iteration
           current_offset += params[:batch_size]
 
           break if result.empty?
 
-          yield result, current_offset if block_given?
+          yield result, current_offset, last_cursor_field_value if block_given?
           # Break the loop if the number of records fetched is less than the batch size
           # break if result.size < params[:batch_size]
         end

server/lib/reverse_etl/utils/cursor_query_builder.rb

@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+module ReverseEtl
+  module Utils
+    class CursorQueryBuilder
+      def self.update_query(sync_config)
+        if sync_config.cursor_field && sync_config.current_cursor_field
+          # If both cursor_field and current_cursor_field are present
+          sync_config.model.query = "(#{existing_query}) AS subquery
+            WHERE #{sync_config.cursor_field} >= '#{sync_config.current_cursor_field}'
+            ORDER BY #{sync_config.cursor_field} ASC"
+        elsif sync_config.cursor_field
+          # If only cursor_field is present but current_cursor_field is not
+          sync_config.model.query = "(#{existing_query}) AS subquery
+            ORDER BY #{sync_config.cursor_field} ASC"
+        end
+      end
+    end
+  end
+end

server/spec/models/sync_spec.rb

@@ -59,11 +59,13 @@
                        })
     end
 
-    let(:sync) { create(:sync, destination:) }
+    let(:sync) { create(:sync, destination:, cursor_field: "cursor_field", current_cursor_field: "2024-01-20") }
 
     it "returns sync config protocol" do
       protocol = sync.to_protocol
       expect(protocol).to be_a(Multiwoven::Integrations::Protocol::SyncConfig)
+      expect(protocol.cursor_field).to eq("cursor_field")
+      expect(protocol.current_cursor_field).to eq("2024-01-20")
     end
   end
 

server/spec/requests/api/v1/syncs_controller_spec.rb

@@ -102,7 +102,8 @@
           sync_interval: 10,
           sync_interval_unit: "minutes",
           stream_name: "profile",
-          sync_mode: "full_refresh"
+          sync_mode: "full_refresh",
+          cursor_field: "created_date"
         }
       }
     end
@@ -127,6 +128,8 @@
         expect(response_hash.dig(:data, :attributes, :model_id)).to eq(request_body.dig(:sync, :model_id))
         expect(response_hash.dig(:data, :attributes, :schedule_type)).to eq(request_body.dig(:sync, :schedule_type))
         expect(response_hash.dig(:data, :attributes, :stream_name)).to eq(request_body.dig(:sync, :stream_name))
+        expect(response_hash.dig(:data, :attributes, :cursor_field)).to eq(request_body.dig(:sync, :cursor_field))
+        expect(response_hash.dig(:data, :attributes, :current_cursor_field)).to eq(nil)
         expect(response_hash.dig(:data, :attributes, :sync_interval_unit))
           .to eq(request_body.dig(:sync, :sync_interval_unit))
         expect(response_hash.dig(:data, :attributes, :sync_interval)).to eq(request_body.dig(:sync, :sync_interval))
@@ -166,7 +169,8 @@
           },
           sync_interval: 10,
           sync_interval_unit: "minutes",
-          stream_name: "profile"
+          stream_name: "profile",
+          cursor_field: "cursor_field"
         }
       }
     end
@@ -188,6 +192,8 @@
         expect(response_hash.dig(:data, :id)).to be_present
         expect(response_hash.dig(:data, :id)).to eq(syncs.first.id.to_s)
         expect(response_hash.dig(:data, :attributes, :sync_interval)).to eq(30)
+        expect(response_hash.dig(:data, :attributes, :cursor_field)).to eq(nil)
+        expect(response_hash.dig(:data, :attributes, :current_cursor_field)).to eq(nil)
       end
 
       it "returns an error response when wrong sync_id" do