Feature/create stream 4 cluster (#93)

* update stream creation parameters

* update test

Author: gangtao

python/test/conftest.py

@@ -4,6 +4,7 @@
 
 import pytest
 import os
+import time
 from timeplus.dbapi import connect
 
 from timeplus import Environment
@@ -33,18 +34,25 @@ def test_stream(test_environment):
 
     try:
         stream.delete()
+        time.sleep(3)
     except Exception:
         pass
 
+    time.sleep(3)
+
     # Create a new stream
     stream = (
         Stream(env=test_environment)
         .name(stream_name)
         .column("time", "integer")
         .column("data", "string")
+        .replication_factor(3)
+        .shards(3)
         .create()
     )
 
+    time.sleep(3)
+
     value = [["time", "data"], [[0, "abcd"]]]
     stream.ingest(*value)
     # Provide the stream to the test

python/test/test_ingest_stream.py

@@ -5,15 +5,19 @@
 from timeplus import Stream, Query
 import datetime
 
+time_wait = 5
 
 def test_ingest(test_environment, test_stream):
+    # there is one row data in test stream already
+    time.sleep(time_wait)
+    # wait previous data ingested
     data = [["time", "data"], [[1, "efgh"]]]
     try:
         test_stream.ingest(*data)
     except Exception as e:
         pytest.fail(f"Ingest method failed with exception {e}")
 
-    time.sleep(3)
+    time.sleep(time_wait)
 
     query = (
         Query(env=test_environment)
@@ -26,7 +30,6 @@ def test_ingest(test_environment, test_stream):
         if event.event == "message":
             results.extend(json.loads(event.data))
     print(results)
-    print(results[1][1])
 
     assert len(results) > 1, "No data returned from the stream"
     assert results[1][0] == 1, "Returned time does not match the ingested integer"
@@ -42,8 +45,12 @@ def test_stream_ingest_lines(test_environment,test_stream):
         Stream(env=test_environment)
         .name("test_stream")
         .column("raw", "string")
+        .replication_factor(3)
+        .shards(3)
         .create()
     )
+    time.sleep(time_wait)
+
     payload = '{"time":1,"data":"abcd"}\n{"time":2,"data":"xyz"}'
 
     # Ingest data in 'lines' format
@@ -52,7 +59,7 @@ def test_stream_ingest_lines(test_environment,test_stream):
     except Exception as e:
         pytest.fail(f"Ingest lines method failed with exception {e}")
 
-    time.sleep(3)
+    time.sleep(time_wait)
 
     query = (
         Query(env=test_environment)
@@ -85,16 +92,20 @@ def test_stream_ingest_raw(test_environment,test_stream):
         Stream(env=test_environment)
         .name("test_stream")
         .column("raw", "string")
+        .replication_factor(3)
+        .shards(3)
         .create()
     )
 
+    time.sleep(time_wait)
+
     # Ingest data in 'raw' format
     try:
         stream.ingest(payload=payload, format="raw")
     except Exception as e:
         pytest.fail(f"Ingest raw method failed with exception {e}")
 
-    time.sleep(3)
+    time.sleep(time_wait)
 
     query = (
         Query(env=test_environment)
@@ -115,6 +126,9 @@ def test_stream_ingest_raw(test_environment,test_stream):
 
 
 def test_json_ingest(test_environment, test_stream):
+    # there is one row data in test stream already
+    time.sleep(time_wait)
+    # wait previous data ingested
     payload = """
     {"time":2,"data":"hello"}
     {"time":1,"data":"world"}
@@ -126,7 +140,7 @@ def test_json_ingest(test_environment, test_stream):
     except Exception as e:
         pytest.fail(f"Ingest streaming method failed with exception {e}")
 
-    time.sleep(3)
+    time.sleep(time_wait)
 
     query = (
         Query(env=test_environment)
@@ -138,8 +152,6 @@ def test_json_ingest(test_environment, test_stream):
         if event.event == "message":
             results.extend(json.loads(event.data))
     print(results)
-    print(results[0][0])
-    print(results[0])
 
     assert len(results) > 1, "No data returned from the stream"
     assert results[1][0] == 2, "Returned data does not match the ingested data"

python/test/test_sqlalchemy_query.py

@@ -1,3 +1,4 @@
+import time
 from sqlalchemy import text, select, MetaData, Table
 from timeplus import View
 
@@ -58,11 +59,30 @@ def test_table_names(engine):
         assert "car_live_data" in tables
 
 
-def test_view_names(engine):
+def test_view_names(test_environment, engine):
+    view_name = "example_mv"
+    view = View(env=test_environment).name(view_name)
+
+    try:
+        view.delete()
+    except Exception:
+        pass
+
+    view = (
+        View(env=test_environment)
+        .name(view_name)
+        .query("select * from test_stream")
+        .create()
+    )
+
+    time.sleep(1)
+
     with engine.connect() as conn:
         views = engine.dialect.get_view_names(conn)
         print(views)
-        assert "car_info" in views
+        assert "example_mv" in views
+
+    view.delete()
 
 
 def test_materialized_view_names(engine,test_environment,test_stream):

python/timeplus/dbapi.py

@@ -3,7 +3,7 @@
 import itertools
 from collections import namedtuple
 
-from timeplus import Environment, Query, Stream, View
+from timeplus import Environment, Query, Stream, View, ExternalStream
 from timeplus.error import Error
 
 
@@ -151,20 +151,27 @@ def __exit__(self, *exc):
     # since some SQL is not exposed through API yet
     def _exist_table(self, name):
         stream = Stream(env=self.env).name(name)
-        return stream.exist()
+        external_stream = ExternalStream(env=self.env).name(name)
+        return stream.exist() or external_stream.exist()
 
     def _exist_view(self, name):
         view = View(env=self.env).name(name)
         return view.exist()
 
     def _get_table(self, name):
-        return Stream(env=self.env).name(name).get()
+        try:
+            return Stream(env=self.env).name(name).get()
+        except:
+            return ExternalStream(env=self.env).name(name).get()
 
     def _get_view(self, name):
         return View(env=self.env).name(name).get()
 
     def _list_table(self):
-        return Stream(env=self.env).list()
+        streams =  Stream(env=self.env).list()
+        estreams = ExternalStream(env=self.env).list()
+
+        return streams + estreams
 
     def _list_view(self):
         return View(env=self.env).list()

python/timeplus/stream.py

@@ -29,6 +29,8 @@ def __init__(self, env):
         self._description = None
         self._mode = "append"
         self._primary_key = None
+        self._replication_factor = None
+        self._shards = None
 
     def name(self, stream_name):
         """
@@ -93,6 +95,14 @@ def primary_key(self, primary_key):
         self._primary_key = primary_key
         return self
 
+    def replication_factor(self, replication_factor):
+        self._replication_factor = replication_factor
+        return self
+
+    def shards(self, shards):
+        self._shards = shards
+        return self
+
     def create(self):
         """
         Sends a request to the API to create the stream.
@@ -125,6 +135,9 @@ def create(self):
         body["description"] = self._description if self._description else None
         body["mode"] = self._mode if self._mode else None
         body["primary_key"] = self._primary_key if self._primary_key else None
+        body["replication_factor"] = self._replication_factor if self._replication_factor else None
+        body["shards"] = self._shards if self._shards else None
+
 
         try:
             self._metadata = self._api_instance.v1beta2_streams_post(body)

python/timeplus/version.py

@@ -1 +1 @@
-__version__ = "1.4.0"  # noqa: F401
+__version__ = "1.4.1"   # noqa: F401