store

Signed-off-by: Ye Cao <[email protected]>

k8s/test/e2e/assembly-demo/assembly-job1.py

@@ -31,7 +31,7 @@ def generate_df(index):
 stream = RecordBatchStream.new(vineyard_client)
 vineyard_client.persist(stream.id)
 print(stream.id)
-writer = stream.writer
+writer = stream.open_writer(vineyard_client)
 total_chunks = 10
 for idx in range(total_chunks):
  time.sleep(idx)

k8s/test/e2e/assembly-demo/assembly-local.py

@@ -24,7 +24,7 @@
 
 stream_id = env_dist['STREAM_ID']
 stream = vineyard_client.get(stream_id)
-reader = stream.reader(vineyard_client)
+reader = stream.open_reader(vineyard_client)
 
 index = 0
 global_meta = vineyard.ObjectMeta()

k8s/test/e2e/assembly-demo/distributed-job1.py

@@ -39,7 +39,7 @@ def generate_df(index):
 
 sys.stdout = sys.__stdout__
 print(meta.id, flush=True)
-writer = stream.writer
+writer = stream.open_writer(vineyard_client)
 total_chunks = 10
 for idx in range(total_chunks):
  time.sleep(idx)

k8s/test/e2e/assembly-demo/distributed-job2.py

@@ -46,7 +46,7 @@ def generate_df(index):
 tup = vineyard_client.create_metadata(meta)
 vineyard_client.persist(tup)
 
-writer = stream.writer
+writer = stream.open_writer(vineyard_client)
 total_chunks = 10
 for idx in range(total_chunks):
  time.sleep(idx)

python/vineyard/io/stream.py

@@ -329,8 +329,8 @@ def meta(self) -> ObjectMeta:
  return self._meta
 
  @property
- def reader(self, client) -> "BaseStream.Reader":
- return self.open_reader(client)
+ def reader(self) -> "BaseStream.Reader":
+ return self.open_reader()
 
  def __str__(self) -> str:
  return repr(self)
@@ -350,8 +350,8 @@ def open_reader(self, client=None) -> "BaseStream.Reader":
  return self._reader
 
  @property
- def writer(self, client) -> "BaseStream.Writer":
- return self.open_writer(client)
+ def writer(self) -> "BaseStream.Writer":
+ return self.open_writer()
 
  def _open_new_writer(self, client) -> "BaseStream.Writer":
  return BaseStream.Writer(client, self.id)