[connector] Tweak ksql for push queries
@@ -111,9 +111,10 @@ class KSqlApi(object):
data = []
metadata = []
- is_select = statement.strip().lower().startswith('select')
- if is_select or statement.strip().lower().startswith('print'):
+ is_select = statement.strip().lower().startswith('select') # TODO via parser
+ is_print = statement.strip().lower().startswith('print')
+ if is_select or is_print:
result = self.client.query(statement)
metadata = [['Row', 'STRING']]
@@ -128,7 +129,14 @@ class KSqlApi(object):
for line in result:
# columns = line.keys()
# data.append([line[col] for col in columns])
- if is_select and line: # Empty first 2 lines?
+ if 'finalMessage' in line:
+ break
+ elif 'header' in line:
+ continue
+ else:
+ line = line.strip()[:-1]
+
+ if is_select:
data_line = json.loads(line)
if data_line.get('@type') == 'statement_error':
raise KSqlApiException(data_line['message'])
