浏览代码

HUE-9077 [ksql] Basic support for DESCRIBE

Romain 6 年之前
父节点
当前提交
e353417f89
共有 1 个文件被更改,包括 13 次插入1 次删除
  1. 13 1
      desktop/libs/kafka/src/kafka/ksql_client.py

+ 13 - 1
desktop/libs/kafka/src/kafka/ksql_client.py

@@ -81,13 +81,14 @@ class KSqlApi(object):
     metadata = []
 
     if statement.strip().lower().startswith('select') or statement.strip().lower().startswith('print'):
-      # STREAMS requires a LIMIT currently or https://github.com/bryanyang0528/ksql-python/pull/60
+      # STREAMS requires a LIMIT currently or will hang without https://github.com/bryanyang0528/ksql-python/pull/60
       result = self.client.query(statement)
       for line in ''.join(list(result)).split('\n'):
         # Until https://github.com/bryanyang0528/ksql-python/issues/57
         # columns = line.keys()
         # data.append([line[col] for col in columns])
         data.append([line])
+        # TODO: WS to plug-in
       metadata = [['Row', 'STRING']]
     else:
       data, metadata = self._decode_result(
@@ -112,6 +113,17 @@ class KSqlApi(object):
             json.dumps(line[column])
           )
         data.append(row)
+    elif result['@type'] == 'sourceDescription':
+      columns = ['name', 'description']
+      for key in result['sourceDescription']:
+        if key == 'fields':
+          data.append(['', ''])
+          data.append(['Fields:', ''])
+          for field in result['sourceDescription'][key]:
+            data.append([field['name'], field['schema']['type']])
+          data.append(['', ''])
+        else:
+          data.append([key, result['sourceDescription'][key]])
 
     columns = [[col, 'STRING'] for col in columns]