[Artifacts] Fix log dataset artifact with df (mlrun#5066)

mlrun/artifacts/manager.py

@@ -132,6 +132,9 @@ def ensure_artifact_source_file_exists(item, path, body):
         #  ModelArtifact is a directory.
         if isinstance(item, ModelArtifact):
             return
+        #  in DatasetArtifact
+        if hasattr(item, "df") and item.df is not None:
+            return
         parsed_url = urlparse(path)
         schema = parsed_url.scheme
         #  we are not checking remote paths yet.

tests/artifacts/test_artifacts.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+import os.path
 import pathlib
 import tempfile
 import typing
@@ -378,6 +379,26 @@ def test_ensure_artifact_source_file_exists(local_path, fail):
             context.log_artifact(item=artifact, local_path=local_path)
 
 
+@pytest.mark.parametrize(
+    "df, fail",
+    [
+        (pd.DataFrame({"num": [0, 1, 2], "color": ["green", "blue", "red"]}), False),
+        (None, True),
+    ],
+)
+def test_ensure_artifact_source_file_exists_by_df(df, fail):
+    context = mlrun.get_or_create_ctx("test")
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+        full_path = os.path.join(temp_dir, "df.parquet")
+        if fail:
+            with pytest.raises(mlrun.errors.MLRunInvalidArgumentError) as error:
+                context.log_dataset(key=str(uuid.uuid4()), df=df, local_path=full_path)
+            assert "Failed to log an artifact, file does not exists" in str(error.value)
+        else:
+            context.log_dataset(key=str(uuid.uuid4()), df=df, local_path=full_path)
+
+
 @pytest.mark.parametrize(
     "artifact,artifact_path,expected_hash,expected_target_path,expected_error",
     [