Forráskód Böngészése

HUE-9077 [ksql] Implement executing a SHOW TOPICS

Romain 6 éve
szülő
commit
1fdbfa11db

+ 37 - 6
desktop/libs/kafka/src/kafka/ksql_client.py

@@ -70,11 +70,42 @@ class KSqlApi(object):
       raise KSqlApiException(e)
 
 
-  def query(self, statement):
-    result = []
+  def ksql(self, statement):
+    response = self.client.ksql(statement)
+    print(response)
+    return response[0]
 
-    for line in self.client.query(statement):
-      data_line = json.loads(line)
-      result.append(data_line['row']['columns']) # TODO: streams, dataline['errorMessage']
 
-    return result
+  def query(self, statement):
+    data = []
+    metadata = []
+
+    if statement.strip().lower().startswith('select'):
+      for line in self.client.query(statement):
+        data_line = json.loads(line)
+        data.append(data_line['row']['columns']) # TODO: streams, dataline['errorMessage']
+    else:
+      data, metadata = self._decode_result(
+        self.ksql(statement)
+      )
+
+    return data, metadata
+
+
+  def _decode_result(self, result):
+    columns = []
+    data = []
+
+    if result['@type'] == 'kafka_topics':
+      columns = result['topics'][0].keys()
+      for line in result['topics']:
+        row = []
+        for column in columns:
+          row.append(
+            json.dumps(line[column])
+          )
+        data.append(row)
+
+    columns = [[col, 'STRING'] for col in columns]
+
+    return data, columns

+ 2 - 0
desktop/libs/notebook/src/notebook/connectors/base.py

@@ -527,6 +527,8 @@ class Api(object):
   def describe_database(self, notebook, snippet, database=None):
     return {}
 
+  def close_statement(self, notebook, snippet): pass
+
   def _get_current_statement(self, notebook, snippet):
     should_close, resp = get_current_statement(snippet)
     if should_close:

+ 1 - 1
desktop/libs/notebook/src/notebook/connectors/ksql.py

@@ -51,7 +51,7 @@ class KSqlApi(Api):
 
   @query_error_handler
   def execute(self, notebook, snippet):
-    data, description = query_and_fetch(self.db, snippet['statement'], 1000)
+    data, description = self.db.query(snippet['statement'])
     has_result_set = data is not None
 
     return {

+ 1 - 0
desktop/libs/notebook/src/notebook/models.py

@@ -495,6 +495,7 @@ class ApiWrapper(object):
   def __init__(self, request, snippet):
     self.request = request
     self.api = _get_api(request, snippet)
+
   def __getattr__(self, name):
     from notebook import tasks as ntasks
     if TASK_SERVER.ENABLED.get() and hasattr(ntasks, name):