Merge pull request #62 from brettshollenberger/fix-deploy-dirs

Fix deploy dirs

Author: brettshollenberger

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    easy_ml (0.2.0.pre.rc95)
+    easy_ml (0.2.0.pre.rc96)
       activerecord
       activerecord-import (~> 1.8.1)
       activesupport

app/models/easy_ml/concerns/versionable.rb

@@ -4,14 +4,23 @@ module Versionable
       extend ActiveSupport::Concern
 
       included do
+        STRING_FORMAT = "%Y_%m_%d_%H_%M_%S".freeze
+
         def bump_version(force: false)
           return version if version.present? && !force
 
-          prev_version = version
-          timestamp = Time.now.utc.strftime("%Y%m%d%H%M%S")
-          timestamp = (prev_version.to_i + 1).to_s if timestamp.to_i <= prev_version.to_i
+          tz = ActiveSupport::TimeZone.new(EasyML::Configuration.timezone)
+          orig_version = version
+          prev_version = tz.parse(version.gsub(/_/, "")) if version
+          timestamp = Time.current.in_time_zone(EasyML::Configuration.timezone)
+          timestamp = (prev_version + 1.second) if prev_version && compare_versions(timestamp, prev_version)
+
+          self.version = timestamp.strftime(STRING_FORMAT)
+        end
 
-          self.version = timestamp
+        def compare_versions(version1, version2)
+          tz = ActiveSupport::TimeZone.new(EasyML::Configuration.timezone)
+          tz.parse(version1.strftime(STRING_FORMAT).gsub(/_/, "")) <= tz.parse(version2.strftime(STRING_FORMAT).gsub(/_/, ""))
         end
       end
     end

app/models/easy_ml/dataset.rb

@@ -114,6 +114,10 @@ def root_dir=(value)
       write_attribute(:root_dir, value)
     end
 
+    def dir
+      root_dir
+    end
+    
     def set_root_dir
       bump_version
       write_attribute(:root_dir, default_root_dir)
@@ -211,12 +215,13 @@ def processed
     end
 
     def bump_versions(version)
+      original_version = self.version
       self.version = version
 
-      @raw = raw.cp(version)
-      @processed = processed.cp(version)
+      @raw = raw.cp(dir.gsub(original_version, version))
+      @processed = processed.cp(dir.gsub(original_version, version))
       save.tap do
-        features.each(&:bump_version)
+        features.each { |feature| feature.bump_version(original_version, version) }
         EasyML::Feature.import(features.to_a, on_duplicate_key_update: [:version])
       end
     end
@@ -750,7 +755,7 @@ def initialize_split(type)
         split_type.new(**args)
       when EasyML::Data::Splits::FileSplit.to_s
         split_type.new(**args.merge(
-                         dir: Pathname.new(root_dir).append("files/splits/#{type}").to_s,
+                         dir: Pathname.new(root_dir).join("files/splits/#{type}").to_s,
                        ))
       end
     end

app/models/easy_ml/feature.rb

@@ -433,9 +433,9 @@ def insert_after(feature_class)
       end
     end
 
-    def bump_version
-      feature_store.bump_version(version)
-      write_attribute(:version, version + 1)
+    def bump_version(original_version, version)
+      feature_store.bump_version(original_version, version)
+      write_attribute(:version, self.version + 1)
       self
     end
 

app/models/easy_ml/model.rb

@@ -203,6 +203,7 @@ def actually_train(&progress_block)
 
     def unlock!
       Support::Lockable.unlock!(lock_key)
+      EasyML::Deploy.new(model: self).unlock!
     end
 
     def lock_model
@@ -558,6 +559,7 @@ def actually_deploy
       dataset.upload_remote_files
       model_snapshot = snapshot
 
+      Thread.current[:deploying] = true
       # Prepare the model to be retrained (reset values so they don't conflict with our snapshotted version)
       bump_version(force: true)
       dataset.bump_versions(version)

easy_ml-0.2.0.pre.rc96.gem

@@ -1,9 +1,10 @@
 module EasyML
   class FeatureStore < EasyML::Data::DatasetManager
-    attr_reader :feature
+    attr_reader :feature, :dataset
 
     def initialize(feature)
       @feature = feature
+      @dataset = feature&.dataset
 
       datasource_config = feature&.dataset&.datasource&.configuration
       if datasource_config
@@ -27,11 +28,11 @@ def synced?
       files.any?
     end
 
-    def bump_version(version)
+    def bump_version(original_version, version)
       compact
       cp(
-        feature_dir_for_version(version),
-        feature_dir_for_version(version + 1),
+        feature_dir.gsub(version, original_version),
+        feature_dir,
       )
     end
 
@@ -41,21 +42,14 @@ def batch_size
       @batch_size ||= feature.batch_size || 10_000
     end
 
-    def feature_dir_for_version(version)
+    def feature_dir
       File.join(
-        Rails.root,
-        "easy_ml/datasets",
-        feature&.dataset&.name&.parameterize&.gsub("-", "_"),
+        dataset.dir,
         "features",
-        feature&.name&.parameterize&.gsub("-", "_"),
-        version.to_s
+        feature&.name&.parameterize&.gsub("-", "_")
       )
     end
 
-    def feature_dir
-      feature_dir_for_version(feature.version)
-    end
-
     def s3_prefix
       File.join("datasets", feature_dir.split("datasets").last)
     end

lib/easy_ml/feature_store.rb

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 module EasyML
-  VERSION = "0.2.0-rc95"
+  VERSION = "0.2.0-rc96"
 
   module Version
   end

lib/easy_ml/version.rb

@@ -51,34 +51,34 @@
       dataset.columns.find_by(name: target).update(is_target: true)
       dataset.columns.where(name: hidden_cols).update_all(hidden: true)
       dataset.columns.find_by(name: "SibSp").update(preprocessing_steps: {
-                                                      training: { method: :median },
-                                                    })
+                                                     training: { method: :median },
+                                                   })
       dataset.columns.find_by(name: "Parch").update(preprocessing_steps: {
-                                                      training: { method: :median },
-                                                    })
+                                                     training: { method: :median },
+                                                   })
       dataset.columns.find_by(name: "Sex").update(preprocessing_steps: {
-                                                    training: {
-                                                      method: :most_frequent,
-                                                      encoding: :one_hot,
-                                                      params: {
-                                                        categorical_min: 2,
-                                                      },
-                                                    },
-                                                  })
+                                                   training: {
+                                                     method: :most_frequent,
+                                                     encoding: :one_hot,
+                                                     params: {
+                                                       categorical_min: 2,
+                                                     },
+                                                   },
+                                                 })
       dataset.columns.find_by(name: "Embarked").update(preprocessing_steps: {
-                                                         training: {
-                                                           method: :most_frequent,
-                                                           encoding: :one_hot,
-                                                           params: {
-                                                             categorical_min: 2,
-                                                           },
-                                                         },
-                                                       })
+                                                        training: {
+                                                          method: :most_frequent,
+                                                          encoding: :one_hot,
+                                                          params: {
+                                                            categorical_min: 2,
+                                                          },
+                                                        },
+                                                      })
       dataset.columns.find_by(name: "Age").update(preprocessing_steps: {
-                                                    training: {
-                                                      method: :median,
-                                                    },
-                                                  })
+                                                   training: {
+                                                     method: :median,
+                                                   },
+                                                 })
       dataset.refresh
     end
   end
@@ -131,15 +131,97 @@ def incr_time
     @time += 1.second
   end
 
+  def relative_dir(dir)
+    dir.split(Rails.root.to_s).last
+  end
+
   describe "#deploy" do
-    it "uses deployed version for prediction" do
+    it "maintains dataset directory structure and versioning" do
+      @t1 = EasyML::Support::EST.parse("2025-01-01").beginning_of_day
+      Timecop.freeze(@t1)
+
       mock_s3_upload
+      model.save
       model.unlock!
 
+      @t2 = EasyML::Support::EST.parse("2025-01-02").beginning_of_day
+      Timecop.freeze(@t2)
+
+      model.train(async: false)
+      model.deploy(async: false)
+      model_v1 = model.current_version
+
+      # Verify initial dataset structure
+      expect(Dir.exist?(model_v1.dataset.raw.dir)).to be true
+      expect(relative_dir(model_v1.dataset.raw.dir)).to eq("/easy_ml/datasets/titanic_dataset/2025_01_01_00_00_00/files/splits/raw")
+
+      expect(Dir.exist?(File.join(model_v1.dataset.dir, "features"))).to be true
+
+      feature_files = model_v1.dataset.features.find_by(name: "Family Size").files
+      expect(feature_files.count).to be > 0
+      feature_files.each do |feature_file|
+        dir = File.dirname(feature_file)
+        expect(relative_dir(dir)).to eq("/easy_ml/datasets/titanic_dataset/2025_01_01_00_00_00/features/family_size/compacted")
+      end
+
+      # New dataset version has been shipped, so it doesn't conflict with the deployed version
+      expect(relative_dir(model.dataset.raw.dir)).to match(%r{/easy_ml/datasets/titanic_dataset/2025_01_02_00_00_\d{2}/files/splits/raw})
+      feature_files = model.dataset.features.find_by(name: "Family Size").files
+      expect(feature_files.count).to be > 0
+      feature_files.each do |feature_file|
+        dir = File.dirname(feature_file)
+        expect(relative_dir(dir)).to match(%r{/easy_ml/datasets/titanic_dataset/2025_01_02_00_00_\d{2}/features/family_size/compacted})
+      end
+
+      # Make changes that require a new version
+      model.dataset.columns.where(name: "Age").update_all(hidden: true)
+      model.dataset.refresh
+
+      @t3 = EasyML::Support::EST.parse("2025-01-03").beginning_of_day
+      Timecop.freeze(@t3)
+
+      model.train(async: false)
+      model.deploy(async: false)
+      model_v2 = model.current_version
+
+      # Verify new version structure
+      expect(Dir.exist?(model_v2.dataset.raw.dir)).to be true
+      expect(Dir.exist?(File.join(model_v2.dataset.dir, "features"))).to be true
+      expect(Dir.exist?(File.join(model_v2.dataset.dir, "features"))).to be true
+
+      # Verify old version files were copied to new version
+      old_files = Dir.glob(File.join(model_v1.dataset.raw.dir, "**/*")).select { |f| File.file?(f) }
+      new_files = Dir.glob(File.join(model_v2.dataset.raw.dir, "**/*")).select { |f| File.file?(f) }
+      expect(old_files.count).to be > 0
+      expect(new_files.count).to be >= old_files.count
+
+      # Test which files are queried
+      #
+      # When using original feature (from v1 model)
+      feature = model_v1.dataset.features.find_by(name: "Family Size")
+      file_pattern = %r{easy_ml/datasets/titanic_dataset/2025_01_01_00_00_\d{2}/features/family_size/compacted/feature.\d.parquet}
+      expect(Polars).to receive(:scan_parquet).with(file_pattern).at_least(:once)
+      feature.query(limit: 1)
+
+      feature_v2 = model_v2.dataset.features.find_by(name: "Family Size")
+      file_pattern_v2 = %r{easy_ml/datasets/titanic_dataset/2025_01_02_00_00_\d{2}/features/family_size/compacted/feature.\d.parquet}
+      expect(Polars).to receive(:scan_parquet).with(file_pattern_v2).at_least(:once)
+      feature_v2.query(limit: 1)
+
+      feature_v3 = model.dataset.features.find_by(name: "Family Size")
+      file_pattern_v3 = %r{easy_ml/datasets/titanic_dataset/2025_01_03_00_00_\d{2}/features/family_size/compacted/feature.\d.parquet}
+      expect(Polars).to receive(:scan_parquet).with(file_pattern_v3).at_least(:once)
+      feature_v3.query(limit: 1)
+
+      Timecop.return
+    end
+
+    it "uses deployed version for prediction" do
+      mock_s3_upload
       @time = EasyML::Support::EST.now
       Timecop.freeze(@time)
 
-      model.save
+      model.save!
       model.unlock!
       model.train(async: false)
       model.deploy(async: false)
@@ -293,7 +375,7 @@ def incr_time
 
       mock_s3_upload
 
-      @time = EasyML::Support::EST.parse("2024-01-01")
+      @time = EasyML::Support::EST.parse("2024-01-01").beginning_of_day
       Timecop.freeze(@time)
 
       model.save
@@ -305,7 +387,7 @@ def incr_time
       model_v1 = model.current_version
 
       def extract_timestamp(dir)
-        EasyML::Support::UTC.parse(dir.gsub(/\D/, "")).in_time_zone(EST)
+        EasyML::Support::EST.parse(dir.gsub(/\D/, ""))
       end
 
       expect(extract_timestamp(model_v1.dataset.raw.dir)).to eq(EasyML::Support::EST.parse("2024-01-01"))