dvc: introduce stage files instead of state files

These describe single stage of pipeline and are allowed to
be placed anywhere in current or parent directories(up to
project's root) of the data file.

By default, 'dvc run' writes stage file to ./stage.dvc, but
it can be changed by using '-f' option(see functest/test-repro.py).

For now we're traversing project tree to find needed stage files for
specific data files. This is very wasteful and not optimal, but will
do for now, as this patch is aimed at stabilizing design of our repo
files.

Signed-off-by: Ruslan Kuprieiev <[email protected]>

dvc/cli.py

@@ -113,6 +113,10 @@ def parse_args(argv=None):
                         action='store_true',
                         default=False,
                         help='Lock data item - disable reproduction.')
+    run_parser.add_argument('-f',
+                        '--file',
+                        default='stage.dvc',
+                        help='Specify name of the state file')
     run_parser.add_argument(
                         'command',
                         nargs=argparse.REMAINDER,

dvc/command/add.py

@@ -3,6 +3,7 @@ import os
 from dvc.command.common.base import CmdBase
 from dvc.logger import Logger
 from dvc.state_file import StateFile
+from dvc.path.data_item import DataItem
 
 
 class CmdAdd(CmdBase):
@@ -39,9 +40,9 @@ class CmdAdd(CmdBase):
         for data_item in targets:
             Logger.debug('Creating state file for {}'.format(data_item.data.relative))
 
-            state_file = StateFile(data_item=data_item,
+            state_file = StateFile(fname=data_item.data.relative + StateFile.STATE_FILE_SUFFIX,
                                    cmd=None,
-                                   out=StateFile.parse_deps_state(self.settings, [data_item.data.dvc]),
+                                   out=StateFile.parse_deps_state(self.settings, [data_item.data.relative], currdir=data_item.data.dirname),
                                    out_git=[],
                                    deps=[],
                                    locked=True)

dvc/command/checkout.py

@@ -35,30 +35,6 @@ class CmdCheckout(CmdBase):
             Logger.info('Checkout \'{}\''.format(item.data.relative))
  
     def run(self):
-        try:
-            states = self.git.state_files_for_previous_commit()
-            prev_items = self.settings.path_factory.data_items_from_states(states, existing=False)
-        except Exception as ex:
-            Logger.error(u'Unable to get data files from previous commit: {}'.format(ex))
-            return 1
-
-        curr_items = self.settings.path_factory.all_existing_data_items()
-        self.checkout(curr_items)
-
-        self.remove_files(list(set(prev_items) - set(curr_items)))
+        items = self.settings.path_factory.all_existing_data_items()
+        self.checkout(items)
         return 0
-
-    @staticmethod
-    def remove_files(removed_items_set):
-        for item in removed_items_set:
-            path = item.data.relative
-            if not os.path.exists(path):
-                Logger.debug(u'File \'{}\' does not exist'.format(path))
-            else:
-                Logger.info(u'Remove \'{}\''.format(path))
-                os.remove(path)
-                dir = os.path.dirname(path)
-                if not os.listdir(dir):
-                    Logger.info(u'Remove directory \'{}\''.format(dir))
-                    os.removedirs(dir)
-        pass

dvc/command/gc.py

@@ -1,49 +1,23 @@
 import os
 
-from dvc.command.common.traverse import Traverse
+from dvc.command.common.base import CmdBase
 from dvc.logger import Logger
-from dvc.path.data_item import DataItemError
 from dvc.config import ConfigI
+from dvc.state_file import StateFile
 
 
-class CmdGC(Traverse):
+class CmdGC(CmdBase):
     def __init__(self, settings):
-        super(CmdGC, self).__init__(settings, "garbage collect", do_not_start_from_root=False)
-        self.clist = []
+        super(CmdGC, self).__init__(settings)
 
     def run(self):
-        if not self._traverse(self.git.git_dir_abs):
-            Logger.error('Failed to collect used cache')
-            return 1
+        clist = [str(x) for x in StateFile.find_all_cache_files(self.git)]
 
         for cache in os.listdir(ConfigI.CACHE_DIR):
             fname = os.path.join(ConfigI.CACHE_DIR, cache)
-            if fname in self.clist:
+            if os.path.basename(fname) in clist:
                 continue
             os.remove(fname)
-            self._remove_cloud_cache(self.settings.path_factory.data_item('.empty', fname))
             Logger.info('Cache \'{}\' was removed'.format(fname))
 
         return 0
-
-    def process_file(self, target):
-        Logger.debug(u'[Cmd-GC] GC file {}.'.format(target))
-
-        try:
-            data_item = self.settings.path_factory.existing_data_item(target)
-        except DataItemError:
-            return
-
-        self.clist.append(data_item.cache.relative)
-        Logger.debug(u'[Cmd-GC] GC data item {}. Success.'.format(data_item.data.relative))
-
-    def is_recursive(self):
-        return self.parsed_args.recursive
-
-    @property
-    def no_git_actions(self):
-        return True
-
-    @staticmethod
-    def not_committed_changes_warning():
-        pass

dvc/command/merge.py

@@ -14,44 +14,35 @@ class CmdMerge(CmdBase):
         super(CmdMerge, self).__init__(settings)
 
     def print_info(self, targets):
-        for item in targets:
+        for fname in targets:
             Logger.info('Restored original data after merge:')
-            Logger.info(' {}'.format(item.data.relative))
-
-    def collect_data(self):
-        dlist = []
-        flist = self.git.get_last_merge_changed_files()
-        for fname in flist:
-            if fname.endswith(DataItem.STATE_FILE_SUFFIX):
-                data = fname[:-len(DataItem.STATE_FILE_SUFFIX)]
-                dlist.append(data)
-        return dlist
+            Logger.info(' {}'.format(fname))
 
     def collect_targets(self):
         targets = []
-        flist = self.collect_data()
-        items = self.settings.path_factory.to_data_items(flist)[0]
 
-        for item in items:
-            command = StateFile.load(item)
+        for fname in self.git.get_last_merge_changed_files():
+            if not StateFile._is_state_file(fname):
+                continue
 
-            if not command.cmd and command.locked:
-                targets.append(item)
+            state = StateFile.load(fname)
+            if not state.cmd and state.locked:
+                targets.append(fname)
 
         return targets
 
     def checkout_targets(self, targets):
-        data = []
-        for item in targets:
-            prev_state = StateFile.loads(self.git.get_file_content_before_last_merge(item.state.relative))
-            curr_state = StateFile.load(item)
+        items = []
+        for fname in targets:
+            self.git.checkout_file_before_last_merge(fname)
+            state = StateFile.load(fname)
+            for out in state.out:
+                item = self.settings.path_factory.data_item(os.path.join(state.cwd, out))
+                items.append(item)
 
-            curr_state.out[item.data.dvc] = prev_state.out[item.data.dvc]
-            curr_state.save()
+        CmdCheckout.checkout(items)
 
-            CmdCheckout.checkout([item])
-
-        msg = 'DVC merge files: {}'.format(' '.join(data))
+        msg = 'DVC merge files: {}'.format(' '.join(targets))
         self.commit_if_needed(msg)
 
     def run(self):

dvc/command/repro.py

@@ -88,7 +88,7 @@ class ReproChange(object):
         self._recursive = recursive
         self._force = force
 
-        self.command = StateFile.load(data_item)
+        self.command = StateFile.find(data_item)
 
         if not self.command.cmd and not self.command.locked:
             msg = 'Error: data item "{}" is not locked, but has no command for reproduction'
@@ -105,15 +105,15 @@ class ReproChange(object):
         return self._cmd_obj
 
     def remove_output_files(self):
-        for output_dvc in self.command.out:
-            Logger.debug('Removing output file {} before reproduction.'.format(output_dvc))
-
+        for out in self.command.out:
+            path = os.path.join(self.command.cwd, out)
+            Logger.debug('Removing output file {} before reproduction.'.format(path))
             try:
-                data_item = self.cmd_obj.settings.path_factory.data_item_from_dvc_path(output_dvc)
+                data_item = self.cmd_obj.settings.path_factory.data_item(path)
                 os.remove(data_item.data.relative)
             except Exception as ex:
                 msg = 'Data item {} cannot be removed before reproduction: {}'
-                Logger.debug(msg.format(output_dvc, ex))
+                Logger.debug(msg.format(path, ex))
 
     def reproduce_run(self):
         Logger.info('Reproducing run command for data item {}. Args: {}'.format(
@@ -169,7 +169,8 @@ class ReproChange(object):
     def reproduce_deps(self, data_item_dvc, recursive):
         result = False
 
-        for path,md5 in self.command.deps.items():
+        for name,md5 in self.command.deps.items():
+            path = os.path.join(self.command.cwd, name)
             if self.reproduce_dep(path, md5, recursive):
                 result = True
 

dvc/command/run.py

@@ -19,44 +19,40 @@ class CmdRun(CmdBase):
 
     def run(self):
         cmd = ' '.join(self.parsed_args.command)
-        state = StateFile(data_item=None,
-                            cmd=cmd,
-                            out=self.parsed_args.out,
-                            out_git=self.parsed_args.out_git,
-                            deps=self.parsed_args.deps,
-                            locked=self.parsed_args.lock)
+
+        if os.path.isfile(self.parsed_args.file):
+            Logger.error("Stage file {} already exists".format(self.parsed_args.file))
+            return 1
+
+        state = StateFile(fname=self.parsed_args.file,
+                          cmd=cmd,
+                          out=self.parsed_args.out,
+                          out_git=self.parsed_args.out_git,
+                          deps=self.parsed_args.deps,
+                          locked=self.parsed_args.lock)
 
         self.run_command(self.settings, state)
         return self.commit_if_needed('DVC run: {}'.format(state.cmd))
 
     @staticmethod
     def run_command(settings, state):
-        Executor.exec_cmd_only_success(state.cmd, shell=True)
+        Executor.exec_cmd_only_success(state.cmd, cwd=state.cwd, shell=True)
 
-        CmdRun.apply_to_files(state.out, state, CmdRun._create_cache_and_state_files, settings)
-        CmdRun.apply_to_files(state.out_git, state, CmdRun._create_state_file, settings)
+        CmdRun.move_output_to_cache(settings, state)
+        CmdRun.update_state_file(settings, state)
 
     @staticmethod
-    def apply_to_files(files, state, func, settings):
-        items = settings.path_factory.to_data_items(files)[0]
-        [func(i, state, settings) for i in items]
+    def update_state_file(settings, state):
+        Logger.debug('Update state file "{}"'.format(state.path))
+        state.out = StateFile.parse_deps_state(settings, state.out)
+        state.out_git = StateFile.parse_deps_state(settings, state.out_git)
+        state.deps = StateFile.parse_deps_state(settings, state.deps)
+        state.save()
 
     @staticmethod
-    def _create_cache_and_state_files(data_item, state, settings):
-        Logger.debug('Move output file "{}" to cache dir "{}" and create a hardlink'.format(
-                     data_item.data.relative, data_item.cache_dir_abs))
-        data_item.move_data_to_cache()
-        return CmdRun._create_state_file(data_item, state, settings)
-
-    @staticmethod
-    def _create_state_file(data_item, state, settings):
-        Logger.debug('Create state file "{}"'.format(data_item.state.relative))
-
-        state_file = StateFile(data_item=data_item,
-                               cmd=state.cmd,
-                               out=StateFile.parse_deps_state(settings, state.out),
-                               out_git=StateFile.parse_deps_state(settings, state.out_git),
-                               locked=state.locked,
-                               deps=StateFile.parse_deps_state(settings, state.deps))
-        state_file.save()
-        return state_file
+    def move_output_to_cache(settings, state):
+        items = settings.path_factory.to_data_items(state.out)[0]
+        for item in items:
+            Logger.debug('Move output file "{}" to cache dir "{}" and create a hardlink'.format(
+                         item.data.relative, item.cache_dir_abs))
+            item.move_data_to_cache()

dvc/path/data_item.py

@@ -33,7 +33,6 @@ class NotInGitDirError(DataDirError):
 
 
 class DataItem(object):
-    STATE_FILE_SUFFIX = '.dvc'
     LOCAL_STATE_FILE_PREFIX = '.'
     LOCAL_STATE_FILE_SUFFIX = '.dvc_local_state'
     CACHE_FILE_SEP = '_'
@@ -78,7 +77,7 @@ class DataItem(object):
 
     @cached_property
     def state(self):
-        return self._state('', self.STATE_FILE_SUFFIX)
+        return Path(StateFile.find(self).path, self._git)
 
     @cached_property
     def cache_dir_abs(self):
@@ -99,7 +98,7 @@ class DataItem(object):
         if self._cache_file:
             file_name = os.path.relpath(os.path.realpath(self._cache_file), cache_dir)
         else:
-            file_name = StateFile.load(self).out[self.data.dvc]
+            file_name = str(StateFile.find_md5(self))
 
         cache_file = os.path.join(cache_dir, file_name)
         return Path(cache_file, self._git)

dvc/path/factory.py

@@ -6,6 +6,7 @@ from dvc.path.data_item import DataItem, DataItemError, DataDirError
 from dvc.path.path import Path
 from dvc.path.stated_data_item import StatedDataItem
 from dvc.system import System
+from dvc.state_file import StateFile
 
 
 class PathFactory(object):
@@ -31,7 +32,8 @@ class PathFactory(object):
             return self.data_item(path.relative)
 
     def is_data_item(self, fname):
-        return os.path.isfile(fname + DataItem.STATE_FILE_SUFFIX)
+        item = DataItem(fname, self._git, self._config)
+        return StateFile.find(item) != None
 
     def existing_data_item(self, fname):
         if not self.is_data_item(fname):
@@ -53,27 +55,5 @@ class PathFactory(object):
         return result, externally_created_files
 
     def all_existing_data_items(self, subdir='.', cache_exists=True):
-        states = []
-
-        for root, dirs, files in os.walk(os.path.join(self._git.git_dir_abs, subdir)):
-            for fname in files:
-                path = os.path.join(root, fname)
-
-                if not fname.endswith(DataItem.STATE_FILE_SUFFIX):
-                    continue
-
-                states.append(path)
-
-        data_items = self.data_items_from_states(states)
-        if cache_exists:
-            data_items = filter(lambda i: os.path.exists(i.state.relative), data_items)
-
-        return data_items
-
-    def data_items_from_states(self, states, existing=True):
-        return [self.data_item_from_dvc_path(self.state_path_to_dvc_path(s), existing)
-                for s in states]
-
-    @staticmethod
-    def state_path_to_dvc_path(state):
-        return re.sub(DataItem.STATE_FILE_SUFFIX + '$', '', state)
+        files = StateFile.find_all_data_files(self._git, subdir)
+        return self.to_data_items(files)[0]

dvc/state_file.py

@@ -3,6 +3,7 @@ import yaml
 
 from dvc.exceptions import DvcException
 from dvc.data_cloud import file_md5
+from dvc.path.path import Path
 
 
 class StateFileError(DvcException):
@@ -24,8 +25,7 @@ class StateFileBase(object):
 
 
 class StateFile(StateFileBase):
-    MAGIC = 'DVC-State'
-    VERSION = '0.1'
+    STATE_FILE_SUFFIX = '.dvc'
 
     PARAM_CMD = 'cmd'
     PARAM_DEPS = 'deps'
@@ -34,14 +34,13 @@ class StateFile(StateFileBase):
     PARAM_OUT_GIT = 'out-git'
 
     def __init__(self,
-                 data_item=None,
+                 fname=None,
                  cmd=None,
                  out=None,
                  out_git=None,
                  deps=None,
                  locked=None):
         super(StateFile, self).__init__()
-        self.data_item = data_item
 
         self.cmd = cmd
         self.out = out
@@ -49,24 +48,30 @@ class StateFile(StateFileBase):
         self.deps = deps
         self.locked = locked
 
+        self.path = os.path.abspath(fname) if fname else None
+        self.cwd = os.path.dirname(self.path) if self.path else None
+
     @staticmethod
-    def parse_deps_state(settings, deps):
+    def parse_deps_state(settings, deps, currdir=None):
         state = {}
         for dep in deps:
             if settings.path_factory.is_data_item(dep):
                 item = settings.path_factory.data_item(dep)
-                state[item.data.dvc] = StateFile.load(item).out[item.data.dvc]
+                md5 = StateFile.find_md5(item)
             else:
-                state[dep] = file_md5(os.path.join(settings.git.git_dir_abs, dep))[0]
+                md5 = file_md5(os.path.join(settings.git.git_dir_abs, dep))[0]
+
+            if currdir:
+                name = os.path.relpath(dep, currdir)
+            else:
+                name = dep
+
+            state[name] = md5
         return state
 
-    @property
-    def file(self):
-        return self.data_item.state.relative
-
     @staticmethod
-    def loadd(data):
-        return StateFile(data_item=None,
+    def loadd(data, fname=None):
+        return StateFile(fname=fname,
                          cmd=data.get(StateFile.PARAM_CMD, None),
                          out=data.get(StateFile.PARAM_OUT, None),
                          out_git=data.get(StateFile.PARAM_OUT_GIT, None),
@@ -74,8 +79,86 @@ class StateFile(StateFileBase):
                          locked=data.get(StateFile.PARAM_LOCKED, None))
 
     @staticmethod
-    def load(data_item):
-        return StateFile._load(data_item.state.relative, StateFile)
+    def load(fname):
+        return StateFile._load(fname, StateFile, fname)
+
+    @staticmethod
+    def _is_state_file(path):
+        return path.endswith(StateFile.STATE_FILE_SUFFIX) and os.path.isfile(path)
+ 
+    @staticmethod
+    def _find_state(fname, dname):
+        name = os.path.relpath(fname, dname)
+        for entry in os.listdir(dname):
+            state_file = os.path.join(dname, entry)
+            if not StateFile._is_state_file(state_file):
+#            if not state_file.endswith(StateFile.STATE_FILE_SUFFIX) or not os.path.isfile(state_file):
+                continue
+            state = StateFile.load(os.path.join(dname, state_file))
+            if name in state.out:
+                return state
+        return None
+
+    @staticmethod
+    def _find(name, start, finish):
+        dname = start
+        fname = name
+        while True:
+            state = StateFile._find_state(fname, dname)
+            if state:
+                return state
+
+            if dname == finish:
+                break
+
+            dname = os.path.dirname(dname)
+        return None
+
+    @staticmethod
+    def find(data_item):
+        return StateFile._find(data_item.data.abs, data_item.data.dirname, data_item._git.git_dir_abs)
+
+    @staticmethod
+    def find_all_state_files(git, subdir='.'):
+        states = []
+        for root, dirs, files in os.walk(os.path.join(git.git_dir_abs, subdir)):        
+            for fname in files:
+                path = os.path.join(root, fname)
+
+                if not StateFile._is_state_file(path):
+                    continue
+
+                states.append(path)
+        return states
+
+    @staticmethod
+    def find_all_states(git, subdir='.'):
+        state_files = StateFile.find_all_state_files(git, subdir)
+        return [StateFile.load(state_file) for state_file in state_files]
+
+    @staticmethod
+    def find_all_data_files(git, subdir='.'):
+        states = StateFile.find_all_states(git, subdir)
+        files = []
+        for state in states:
+            for out in state.out:
+                files.append(os.path.join(state.cwd, out))
+        return files
+
+    @staticmethod
+    def find_all_cache_files(git, subdir='.'):
+        states = StateFile.find_all_states(git, subdir)
+        cache_files = []
+        for state in states:
+            for out,md5 in state.out.items():
+                cache_files.append(md5)
+        return cache_files
+
+    @staticmethod
+    def find_md5(data_item):
+        state = StateFile.find(data_item)
+        name = os.path.relpath(data_item.data.abs, os.path.dirname(state.path))
+        return state.out[name]
 
     @staticmethod
     def loads(content):
@@ -91,11 +174,7 @@ class StateFile(StateFileBase):
             self.PARAM_LOCKED: self.locked
         }
 
-        file_dir = os.path.dirname(self.file)
-        if file_dir != '' and not os.path.isdir(file_dir):
-            os.makedirs(file_dir)
-
-        self._save(self.file, res)
+        self._save(self.path, res)
 
 
 class LocalStateFile(StateFileBase):

dvc/utils.py

@@ -79,7 +79,7 @@ def map_progress(func, targets, n_threads):
     try:
         ret = pool.map(func, targets)
     except Exception as exc:
-        Logger.error('Unexpected exception while processing targets: {}'.format(exc))
+        Logger.error('Unexpected exception while processing targets: {}'.format(exc), exc_info=True)
     finally:
         progress.finish()
 

functest/test-repro-code.sh

@@ -10,7 +10,7 @@ function test_generic() {
 	dvc_create_repo
 
 	dvc_info 'Copy foo into foo1'
-	dvc run -d code/code.sh -d data/foo -o data/foo1 bash code/code.sh data/foo data/foo1
+	dvc run -f copy_foo_foo1.dvc -d code/code.sh -d data/foo -o data/foo1 bash code/code.sh data/foo data/foo1
 
 	dvc_info 'Modify code'
 	echo " " >> code/code.sh 
@@ -52,9 +52,9 @@ function test_partial() {
 	git commit -m "copy code"
 
 	dvc_info "Create repro chain foo -> foo1 -> foo2 -> foo3"
-	dvc run -d code/code1.sh -d data/foo -o data/foo1 cp data/foo data/foo1
-	dvc run -d code/code2.sh -d data/foo1 -o data/foo2 cp data/foo1 data/foo2
-	dvc run -d code/code3.sh -d data/foo2 -o data/foo3 cp data/foo2 data/foo3
+	dvc run -f copy_foo_foo1.dvc -d code/code1.sh -d data/foo -o data/foo1 cp data/foo data/foo1
+	dvc run -f copy_foo1_foo2.dvc -d code/code2.sh -d data/foo1 -o data/foo2 cp data/foo1 data/foo2
+	dvc run -f copy_foo2_foo3.dvc -d code/code3.sh -d data/foo2 -o data/foo3 cp data/foo2 data/foo3
 
 	dvc_info "Save original timestamps"
 	FOO_TS=$(dvc_timestamp data/foo)