feat(executors): add support for specifying prefix for query path

Author: AndreyMarkinPPC

libs/executors/garf/executors/entrypoints/cli.py

@@ -105,12 +105,14 @@ def main():
           )
         for query in queries:
           if isinstance(query, garf.executors.workflow.QueryPath):
+            query_path = query.full_path
             if re.match(
-              '^(http|gs|s3|aruze|hdfs|webhdfs|ssh|scp|sftp)', query.path
+              '^(http|gs|s3|aruze|hdfs|webhdfs|ssh|scp|sftp)', query_path
             ):
-              batch[query.path] = reader_client.read(query.path)
+              batch[query.path] = reader_client.read(query_path)
             else:
-              query_path = wf_parent / pathlib.Path(query.path)
+              if not query.prefix:
+                query_path = wf_parent / pathlib.Path(query.path)
               if not query_path.exists():
                 raise workflow.GarfWorkflowError(
                   f'Query: {query_path} not found'

libs/executors/garf/executors/workflow.py

@@ -15,6 +15,7 @@
 
 import os
 import pathlib
+import re
 
 import pydantic
 import smart_open
@@ -37,6 +38,13 @@ class QueryPath(pydantic.BaseModel):
   """Path file with query."""
 
   path: str
+  prefix: str | None = None
+
+  @property
+  def full_path(self) -> str:
+    if self.prefix:
+      return re.sub('/$', '', self.prefix) + '/' + self.path
+    return self.path
 
 
 class QueryDefinition(pydantic.BaseModel):