Merge pull request #2384 from dhruvesh09/r0.23.0

Use `component/{run_id}/value` instead of `component/{run_id}` as fil…

Author: dhruvesh09

tfx/components/base/base_driver.py

@@ -52,6 +52,8 @@ def _prepare_output_paths(artifact: types.Artifact):
     # idempotent executions is needed.
     return
 
+  # TODO(b/147242148): Introduce principled artifact structure (directory
+  # or file) definition.
   if isinstance(artifact, types.ValueArtifact):
     artifact_dir = os.path.dirname(artifact.uri)
   else:
@@ -222,6 +224,10 @@ def _prepare_output_artifacts(
       for i, artifact in enumerate(output_list):
         artifact.uri = _generate_output_uri(base_output_dir, name, execution_id,
                                             is_single_artifact, i)
+        # TODO(b/147242148): Introduce principled artifact structure (directory
+        # or file) definition.
+        if isinstance(artifact, types.ValueArtifact):
+          artifact.uri = os.path.join(artifact.uri, 'value')
         _prepare_output_paths(artifact)
 
       result[name] = output_list

tfx/orchestration/portable/outputs_utils.py

@@ -63,6 +63,7 @@ def __init__(self, pipeline_node: pipeline_pb2.PipelineNode,
   def generate_output_artifacts(
       self, execution_id: int) -> Dict[Text, List[types.Artifact]]:
     """Generates output artifacts given execution_id."""
+    # TODO(b/166312308): support ValueArtifact behavior.
     output_artifacts = collections.defaultdict(list)
     for key, output_spec in self._pipeline_node.outputs.outputs.items():
       artifact = types.Artifact(

tfx/types/artifact_test.py

@@ -87,7 +87,7 @@ def decode(self, value: bytes):
 _BYTE_VALUE = b'This is a string'
 
 # Mock paths for string artifact.
-_VALID_URI = '/tmp/uri'
+_VALID_URI = '/tmp/uri/value'
 _VALID_FILE_URI = _VALID_URI
 
 # Mock invalid paths. _BAD_URI points to a valid dir but there's no file within.