Merge branch 'dev-2.0.0-beta' of https://github.com/FederatedAI/FATE …

…into feature-2.0.0-glm # Conflicts: # python/fate/components/components/__init__.py # python/fate/components/components/hetero_lr.py # python/fate/ml/lr/arbiter.py # python/fate/ml/lr/guest.py # python/fate/ml/lr/host.py
FederatedAI · Jun 28, 2023 · f6297a0 · f6297a0
2 parents 0c7b841 + e781500
commit f6297a0
Show file tree

Hide file tree

Showing 131 changed files with 3,826 additions and 4,473 deletions.
diff --git a/doc/2.0/quick_start.md b/doc/2.0/quick_start.md
@@ -34,7 +34,7 @@ host_data_path = os.path.join(base_path, "breast_hetero_host.csv")
 # create pipeline
 pipeline = StandalonePipeline().set_roles(guest="9999", host="10000", arbiter="10001")
 
-# create reader component
+# create reader component_desc
 reader_0 = Reader(name="reader_0")
 reader_0.guest.component_param(
     path=f"file://${guest_data_path}",
@@ -54,11 +54,11 @@ reader_0.hosts[0].component_param(
     dtype="float32",
 )
 
-# create intersection component
+# create intersection component_desc
 intersection_0 = Intersection(name="intersection_0", method="raw", input_data=reader_0.outputs["output_data"])
 intersection_1 = Intersection(name="intersection_1", method="raw", input_data=reader_0.outputs["output_data"])
 
-# create feature scale component
+# create feature scale component_desc
 feature_scale_0 = FeatureScale(
     name="feature_scale_0", method="standard", train_data=intersection_0.outputs["output_data"]
 )
@@ -68,7 +68,7 @@ feature_scale_1 = FeatureScale(
     input_model=feature_scale_0.outputs["output_model"],
 )
 
-# create lr component
+# create lr component_desc
 lr_0 = HeteroLR(
     name="lr_0",
     train_data=feature_scale_0.outputs["train_output_data"],
@@ -78,7 +78,7 @@ lr_0 = HeteroLR(
     batch_size=-1,
 )
 
-# create evaluation component
+# create evaluation component_desc
 evaluation_0 = Evaluation(name="evaluation_0", runtime_roles="guest", input_data=lr_0.outputs["train_output_data"])
 
 # add components

diff --git a/python/fate/arch/__init__.py b/python/fate/arch/__init__.py
@@ -15,6 +15,6 @@
 #
 
 from .context import Context
-from .unify import Backend, device
+from .unify import URI, Backend, device
 
-__all__ = ["Backend", "device", "Context"]
+__all__ = ["Backend", "device", "Context", "URI"]
diff --git a/python/fate/arch/computing/_address.py b/python/fate/arch/computing/_address.py
diff --git a/python/fate/arch/computing/_computing.py b/python/fate/arch/computing/_computing.py
diff --git a/python/fate/arch/computing/eggroll/_csession.py b/python/fate/arch/computing/eggroll/_csession.py
@@ -15,11 +15,10 @@
 #
 
 import logging
-from typing import Optional
 
 from fate.interface import CSessionABC
 
-from ...unify import uuid
+from ...unify import URI, uuid
 from .._profile import computing_profile
 from ._table import Table
 
@@ -52,35 +51,37 @@ def session_id(self):
         return self._session_id
 
     @computing_profile
-    def load(self, address, partitions: Optional[int], schema: dict, **kwargs):
-
-        from .._address import EggRollAddress
+    def load(self, uri: URI, schema: dict, options: dict = None) -> Table:
         from ._type import EggRollStoreType
 
-        if isinstance(address, EggRollAddress):
-            options = kwargs.get("option", {})
-            if partitions is not None:
-                options["total_partitions"] = partitions
-            options["store_type"] = kwargs.get("store_type", EggRollStoreType.ROLLPAIR_LMDB)
-            options["create_if_missing"] = False
-            rp = self._rpc.load(namespace=address.namespace, name=address.name, options=options)
-            if rp is None or rp.get_partitions() == 0:
-                raise RuntimeError(f"no exists: {address.name}, {address.namespace}")
-
-            if options["store_type"] != EggRollStoreType.ROLLPAIR_IN_MEMORY:
-                rp = rp.save_as(
-                    name=f"{address.name}_{uuid()}",
-                    namespace=self.session_id,
-                    partition=partitions,
-                    options={"store_type": EggRollStoreType.ROLLPAIR_IN_MEMORY},
-                )
+        if uri.schema != "eggroll":
+            raise ValueError(f"uri scheme {uri.schema} not supported with eggroll backend")
+        try:
+            _, namespace, name = uri.path_splits()
+        except Exception as e:
+            raise ValueError(f"uri {uri} not valid, demo format: eggroll:///namespace/name") from e
+
+        if options is None:
+            options = {}
+        if "store_type" not in options:
+            options["store_type"] = EggRollStoreType.ROLLPAIR_LMDB
+        options["create_if_missing"] = False
+        rp = self._rpc.load(namespace=namespace, name=name, options=options)
+        if rp is None or rp.get_partitions() == 0:
+            raise RuntimeError(f"no exists: {name}, {namespace}")
+
+        if options["store_type"] != EggRollStoreType.ROLLPAIR_IN_MEMORY:
+            rp = rp.save_as(
+                name=f"{name}_{uuid()}",
+                namespace=self.session_id,
+                partition=rp.get_partitions(),
+                options={"store_type": EggRollStoreType.ROLLPAIR_IN_MEMORY},
+            )
 
             table = Table(rp=rp)
             table.schema = schema
             return table
 
-        raise NotImplementedError(f"address type {type(address)} not supported with eggroll backend")
-
     @computing_profile
     def parallelize(self, data, partition: int, include_key: bool, **kwargs) -> Table:
         options = dict()