You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "vtlim (via GitHub)" <gi...@apache.org> on 2023/02/24 00:21:50 UTC
[GitHub] [druid] vtlim commented on a diff in pull request #13787: Python Druid API for use in notebooks
vtlim commented on code in PR #13787:
URL: https://github.com/apache/druid/pull/13787#discussion_r1116298038
##########
examples/quickstart/jupyter-notebooks/druidapi/datasource.py:
##########
@@ -0,0 +1,111 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests, time
+from .consts import COORD_BASE
+from .rest import check_error
+from .util import dict_get
+
+REQ_DATASOURCES = COORD_BASE + '/datasources'
+REQ_DATASOURCE = REQ_DATASOURCES + '/{}'
+
+# Segment load status
+REQ_DATASOURCES = COORD_BASE + '/datasources'
+REQ_DS_LOAD_STATUS = REQ_DATASOURCES + '/{}/loadstatus'
+
+class DatasourceClient:
+ '''
+ Client for status APIs. These APIs are available on all nodes.
+ If used with the router, they report the status of just the router.
+ '''
+
+ def __init__(self, rest_client):
+ self.rest_client = rest_client
+
+ def names(self, include_unused=False, include_disabled=False):
+ """
+ Returns a list of the names of data sources in the metadata store.
+
+ Parameters
+ ----------
+ include_unused : bool, default = False
+ if False, returns only datasources with at least one used segment
+ in the cluster.
+
+ include_unused : bool, default = False
+ if False, returns only enamed datasources.
Review Comment:
```suggestion
include_disabled : bool, default = False
If False, returns only enabled datasources.
```
Not sure if this is what you meant or what an "enabled datasource" is
##########
examples/quickstart/jupyter-notebooks/druidapi/display.py:
##########
@@ -0,0 +1,84 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+TEXT_TABLE = 0
+HTML_TABLE = 1
+
+class Display:
+
+ def __init__(self):
+ self.format = TEXT_TABLE
+ self.html_initialized = False
+
+ def text(self):
+ self.format = TEXT_TABLE
+
+ def html(self):
+ self.format = HTML_TABLE
+ if not self.html_initialized:
+ from .html_table import styles
+ styles()
+ self.html_initialized = True
+
+ def table(self):
+ if self.format == HTML_TABLE:
+ from .html_table import HtmlTable
+ return HtmlTable()
+ else:
+ from .text_table import TextTable
+ return TextTable()
+
+ def show_object_list(self, objects, cols):
+ list_to_table(self.table(), objects, cols)
+
+ def show_object(self, obj, labels):
+ object_to_table(self.table(), obj, labels)
+
+ def show_error(self, msg):
+ from .html_table import html_error
+ html_error("<b>ERROR: " + msg + "</b")
+
+ def show_message(self, msg):
+ from .html_table import html
+ html("<b>" + msg + "</b")
Review Comment:
Missing closing bracket?
##########
examples/quickstart/jupyter-notebooks/druidapi/html_table.py:
##########
@@ -0,0 +1,119 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from IPython.display import display, HTML
+from .base_table import BaseTable
+from html import escape
+
+STYLES = '''
+<style>
+ .druid table {
+ border: 1px solid black;
+ border-collapse: collapse;
+ }
+
+ .druid th, .druid td {
+ padding: 4px 1em ;
+ text-align: left;
+ }
+
+ td.druid-right, th.druid-right {
+ text-align: right;
+ }
+
+ td.druid-center, th.druid-center {
+ text-align: center;
+ }
+
+ .druid .druid-left {
+ text-align: left;
+ }
+
+ .druid-alert {
+ color: red;
+ }
+</style>
+'''
+
+def escape_for_html(s):
+ # Anoying: IPython treats $ as the start of Latex, which is cool,
Review Comment:
```suggestion
# Annoying: IPython treats $ as the start of Latex, which is cool,
```
##########
examples/quickstart/jupyter-notebooks/druidapi/rest.py:
##########
@@ -0,0 +1,180 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests
+from .util import dict_get, is_blank
+from urllib.parse import quote
+from .error import ClientError
+
+def check_error(response):
+ """
+ Raises a requests HttpError if the response code is not OK or Accepted.
+
+ If the response included a JSON payload, then the message is extracted
+ from that payload, else the message is from requests. The JSON
+ payload, if any, is returned in the json field of the error.
+ """
+ code = response.status_code
+ if code == requests.codes.ok or code == requests.codes.accepted:
+ return
+ error = None
+ json = None
+ try:
+ json = response.json()
+ except Exception:
+ # If we can't get the JSON, just move on, we'll figure
+ # things out another way.
+ pass
+ msg = dict_get(json, 'errorMessage')
+ if msg is None:
+ msg = dict_get(json, 'error')
+ if not is_blank(msg):
+ raise ClientError(msg)
+ if code == requests.codes.not_found and error is None:
+ error = "Not found"
+ if error is not None:
+ response.reason = error
+ try:
+ response.raise_for_status()
+ except Exception as e:
+ e.json = json
+ raise e
+
+class DruidRestClient:
+ '''
+ Wrapper around the basic Druid REST API operations using the
+ requests Python package. Handles the grunt work of building up
+ URLs, working with JSON, etc.
+ '''
+
+ def __init__(self, endpoint):
+ self.endpoint = endpoint
+ self.trace = False
+ self.session = requests.Session()
+
+ def enable_trace(self, flag=True):
+ self.trace = flag
+
+ def build_url(self, req, args=None) -> str:
+ """
+ Returns the full URL for a REST call given the relative request API and
+ optional parameters to fill placeholders within the request URL.
+
+ Parameters
+ ----------
+ req : str
+ relative URL, with optional {} placeholders
Review Comment:
```suggestion
Relative URL, with optional {} placeholders.
Example:`/status`
```
##########
examples/quickstart/jupyter-notebooks/druidapi/catalog.py:
##########
@@ -0,0 +1,60 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests
+from .consts import COORD_BASE
+from .rest import check_error
+
+# Catalog (new feature in Druid 26)
+CATALOG_BASE = COORD_BASE + '/catalog'
+REQ_CAT_SCHEMAS = CATALOG_BASE + '/schemas'
+REQ_CAT_SCHEMA = REQ_CAT_SCHEMAS + '/{}'
+REQ_CAT_SCHEMA_TABLES = REQ_CAT_SCHEMA + '/tables'
+REQ_CAT_SCHEMA_TABLE = REQ_CAT_SCHEMA_TABLES + '/{}'
+REQ_CAT_SCHEMA_TABLE_EDIT = REQ_CAT_SCHEMA_TABLE + '/edit'
+
+class CatalogClient:
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def post_table(self, schema, table_name, table_spec, version=None, overwrite=None):
+ params = {}
+ if version is not None:
+ params['version'] = version
+ if overwrite is not None:
+ params['overwrite'] = overwrite
+ return self.client.post_json(REQ_CAT_SCHEMA_TABLE, table_spec, args=[schema, table_name], params=params)
+
+ def create(self, schema, table_name, table_spec):
+ self.post_table(schema, table_name, table_spec)
+
+ def table(self, schema, table_name):
+ return self.client.get_json(REQ_CAT_SCHEMA_TABLE, args=[schema, table_name])
+
+ def drop_table(self, schema, table_name, ifExists=False):
Review Comment:
nit: stick with snake case for consistency
##########
examples/quickstart/jupyter-notebooks/druidapi/sql.py:
##########
@@ -0,0 +1,690 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time, requests
+from . import consts, display
+from .consts import ROUTER_BASE
+from .util import is_blank, dict_get
+from .error import DruidError, ClientError
+
+REQ_ROUTER_QUERY = ROUTER_BASE
+REQ_ROUTER_SQL = ROUTER_BASE + '/sql'
+REQ_ROUTER_SQL_TASK = REQ_ROUTER_SQL + '/task'
+
+class SqlRequest:
+
+ def __init__(self, query_client, sql):
+ self.query_client = query_client
+ self.sql = sql
+ self.context = None
+ self.params = None
+ self.header = False
+ self.format = consts.SQL_OBJECT
+ self.headers = None
+ self.types = None
+ self.sqlTypes = None
+
+ def with_format(self, result_format):
+ self.format = result_format
+ return self
+
+ def with_headers(self, sqlTypes=False, druidTypes=False):
+ self.headers = True
+ self.types = druidTypes
+ self.sqlTypes = sqlTypes
+ return self
+
+ def with_context(self, context):
+ if self.context is None:
+ self.context = context
+ else:
+ self.context.update(context)
+ return self
+
+ def with_parameters(self, params):
+ '''
+ Set the array of parameters. Parameters must each be a map of 'type'/'value' pairs:
+ {'type': the_type, 'value': the_value}. The type must be a valid SQL type
+ (in upper case). See the consts module for a list.
+ '''
+ if self.params is None:
+ self.params = params
+ else:
+ self.params.update(params)
+ return self
+
+ def add_parameter(self, value):
+ '''
+ Add one parameter value. Infers the type of the parameter from the Python type.
+ '''
+ if value is None:
+ raise ClientError("Druid does not support null parameter values")
+ data_type = None
+ value_type = type(value)
+ if value_type is str:
+ data_type = consts.SQL_VARCHAR_TYPE
+ elif value_type is int:
+ data_type = consts.SQL_BIGINT_TYPE
+ elif value_type is float:
+ data_type = consts.SQL_DOUBLE_TYPE
+ elif value_type is list:
+ data_type = consts.SQL_ARRAY_TYPE
+ else:
+ raise ClientError("Unsupported value type")
+ if self.params is None:
+ self.params = []
+ self.params.append({'type': data_type, 'value': value})
Review Comment:
note that you use a mix of single quotes and double quotes throughout the library. for instance, compare this line with L99
##########
examples/quickstart/jupyter-notebooks/druidapi/datasource.py:
##########
@@ -0,0 +1,111 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests, time
+from .consts import COORD_BASE
+from .rest import check_error
+from .util import dict_get
+
+REQ_DATASOURCES = COORD_BASE + '/datasources'
+REQ_DATASOURCE = REQ_DATASOURCES + '/{}'
+
+# Segment load status
+REQ_DATASOURCES = COORD_BASE + '/datasources'
+REQ_DS_LOAD_STATUS = REQ_DATASOURCES + '/{}/loadstatus'
+
+class DatasourceClient:
+ '''
+ Client for status APIs. These APIs are available on all nodes.
+ If used with the router, they report the status of just the router.
+ '''
+
+ def __init__(self, rest_client):
+ self.rest_client = rest_client
+
+ def names(self, include_unused=False, include_disabled=False):
+ """
+ Returns a list of the names of data sources in the metadata store.
+
+ Parameters
+ ----------
+ include_unused : bool, default = False
+ if False, returns only datasources with at least one used segment
+ in the cluster.
+
+ include_unused : bool, default = False
+ if False, returns only enamed datasources.
+
+ Reference
+ ---------
+ * `GET /druid/coordinator/v1/metadata/datasources`
+ * `GET /druid/coordinator/v1/metadata/datasources?includeUnused`
+ * `GET /druid/coordinator/v1/metadata/datasources?includeDisabled`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-4
+
+ Note: this method uses a semi-deprecated API.
+ See `Metadata.user_table_names()` for the preferred solution.
+ """
+ params = {}
+ if include_unused:
+ params['includeUnused'] = ''
+ if include_disabled:
+ params['includeDisabled'] = ''
+ return self.rest_client.get_json(REQ_DATASOURCES, params=params)
+
+ def drop(self, ds_name, ifExists=False):
+ """
+ Drops a data source.
+
+ Marks as unused all segments belonging to a datasource.
+
+ Marking all segments as unused is equivalent to dropping the table.
+
+ Parameters
+ ----------
+ ds_name: str
+ name of the datasource to query
Review Comment:
```suggestion
ds_name : str
Name of the datasource to query.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/sql.py:
##########
@@ -0,0 +1,690 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time, requests
+from . import consts, display
+from .consts import ROUTER_BASE
+from .util import is_blank, dict_get
+from .error import DruidError, ClientError
+
+REQ_ROUTER_QUERY = ROUTER_BASE
+REQ_ROUTER_SQL = ROUTER_BASE + '/sql'
+REQ_ROUTER_SQL_TASK = REQ_ROUTER_SQL + '/task'
+
+class SqlRequest:
+
+ def __init__(self, query_client, sql):
+ self.query_client = query_client
+ self.sql = sql
+ self.context = None
+ self.params = None
+ self.header = False
+ self.format = consts.SQL_OBJECT
+ self.headers = None
+ self.types = None
+ self.sqlTypes = None
+
+ def with_format(self, result_format):
+ self.format = result_format
+ return self
+
+ def with_headers(self, sqlTypes=False, druidTypes=False):
+ self.headers = True
+ self.types = druidTypes
+ self.sqlTypes = sqlTypes
+ return self
+
+ def with_context(self, context):
+ if self.context is None:
+ self.context = context
+ else:
+ self.context.update(context)
+ return self
+
+ def with_parameters(self, params):
+ '''
+ Set the array of parameters. Parameters must each be a map of 'type'/'value' pairs:
+ {'type': the_type, 'value': the_value}. The type must be a valid SQL type
+ (in upper case). See the consts module for a list.
+ '''
+ if self.params is None:
+ self.params = params
+ else:
+ self.params.update(params)
+ return self
+
+ def add_parameter(self, value):
+ '''
+ Add one parameter value. Infers the type of the parameter from the Python type.
+ '''
+ if value is None:
+ raise ClientError("Druid does not support null parameter values")
+ data_type = None
+ value_type = type(value)
+ if value_type is str:
+ data_type = consts.SQL_VARCHAR_TYPE
+ elif value_type is int:
+ data_type = consts.SQL_BIGINT_TYPE
+ elif value_type is float:
+ data_type = consts.SQL_DOUBLE_TYPE
+ elif value_type is list:
+ data_type = consts.SQL_ARRAY_TYPE
+ else:
+ raise ClientError("Unsupported value type")
+ if self.params is None:
+ self.params = []
+ self.params.append({'type': data_type, 'value': value})
+
+ def response_header(self):
+ self.header = True
+ return self
+
+ def request_headers(self, headers):
+ self.headers = headers
+ return self
+
+ def to_request(self):
+ query_obj = {"query": self.sql}
+ if self.context is not None and len(self.context) > 0:
+ query_obj['context'] = self.context
+ if self.params is not None and len(self.params) > 0:
+ query_obj['parameters'] = self.params
+ if self.header:
+ query_obj['header'] = True
+ if self.result_format is not None:
+ query_obj['resultFormat'] = self.format
+ if self.sqlTypes:
+ query_obj['sqlTypesHeader'] = self.sqlTypes
+ if self.types:
+ query_obj['typesHeader'] = self.types
+ return query_obj
+
+ def result_format(self):
+ return self.format.lower()
+
+ def run(self):
+ return self.query_client.sql_query(self)
+
+def parse_rows(fmt, context, results):
+ if fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ rows = results['results']
+ elif fmt == consts.SQL_ARRAY:
+ rows = results
+ else:
+ return results
+ if not context.get('headers', False):
+ return rows
+ header_size = 1
+ if context.get('sqlTypesHeader', False):
+ header_size += 1
+ if context.get('typesHeader', False):
+ header_size += 1
+ return rows[header_size:]
+
+def label_non_null_cols(results):
+ if results is None or len(results) == 0:
+ return []
+ is_null = {}
+ for key in results[0].keys():
+ is_null[key] = True
+ for row in results:
+ for key, value in row.items():
+ if type(value) == str:
+ if value != '':
+ is_null[key] = False
+ elif type(value) == float:
+ if value != 0.0:
+ is_null[key] = False
+ elif value is not None:
+ is_null[key] = False
Review Comment:
Would it make sense to check "if not None" first? Because if that's true, it won't need to through the logic for "if string" or "if float".
##########
examples/quickstart/jupyter-notebooks/druidapi/datasource.py:
##########
@@ -0,0 +1,111 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests, time
+from .consts import COORD_BASE
+from .rest import check_error
+from .util import dict_get
+
+REQ_DATASOURCES = COORD_BASE + '/datasources'
+REQ_DATASOURCE = REQ_DATASOURCES + '/{}'
+
+# Segment load status
+REQ_DATASOURCES = COORD_BASE + '/datasources'
+REQ_DS_LOAD_STATUS = REQ_DATASOURCES + '/{}/loadstatus'
+
+class DatasourceClient:
+ '''
+ Client for status APIs. These APIs are available on all nodes.
+ If used with the router, they report the status of just the router.
+ '''
Review Comment:
nit: stay consistent with using single or double quotes for triple quotes
```suggestion
"""
Client for status APIs. These APIs are available on all nodes.
If used with the Router, they report the status of just the Router.
"""
```
##########
examples/quickstart/jupyter-notebooks/druidapi/sql.py:
##########
@@ -0,0 +1,690 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time, requests
+from . import consts, display
+from .consts import ROUTER_BASE
+from .util import is_blank, dict_get
+from .error import DruidError, ClientError
+
+REQ_ROUTER_QUERY = ROUTER_BASE
+REQ_ROUTER_SQL = ROUTER_BASE + '/sql'
+REQ_ROUTER_SQL_TASK = REQ_ROUTER_SQL + '/task'
+
+class SqlRequest:
+
+ def __init__(self, query_client, sql):
+ self.query_client = query_client
+ self.sql = sql
+ self.context = None
+ self.params = None
+ self.header = False
+ self.format = consts.SQL_OBJECT
+ self.headers = None
+ self.types = None
+ self.sqlTypes = None
+
+ def with_format(self, result_format):
+ self.format = result_format
+ return self
+
+ def with_headers(self, sqlTypes=False, druidTypes=False):
+ self.headers = True
+ self.types = druidTypes
+ self.sqlTypes = sqlTypes
+ return self
+
+ def with_context(self, context):
+ if self.context is None:
+ self.context = context
+ else:
+ self.context.update(context)
+ return self
+
+ def with_parameters(self, params):
+ '''
+ Set the array of parameters. Parameters must each be a map of 'type'/'value' pairs:
+ {'type': the_type, 'value': the_value}. The type must be a valid SQL type
+ (in upper case). See the consts module for a list.
+ '''
+ if self.params is None:
+ self.params = params
+ else:
+ self.params.update(params)
+ return self
+
+ def add_parameter(self, value):
+ '''
+ Add one parameter value. Infers the type of the parameter from the Python type.
+ '''
+ if value is None:
+ raise ClientError("Druid does not support null parameter values")
+ data_type = None
+ value_type = type(value)
+ if value_type is str:
+ data_type = consts.SQL_VARCHAR_TYPE
+ elif value_type is int:
+ data_type = consts.SQL_BIGINT_TYPE
+ elif value_type is float:
+ data_type = consts.SQL_DOUBLE_TYPE
+ elif value_type is list:
+ data_type = consts.SQL_ARRAY_TYPE
+ else:
+ raise ClientError("Unsupported value type")
+ if self.params is None:
+ self.params = []
+ self.params.append({'type': data_type, 'value': value})
+
+ def response_header(self):
+ self.header = True
+ return self
+
+ def request_headers(self, headers):
+ self.headers = headers
+ return self
+
+ def to_request(self):
+ query_obj = {"query": self.sql}
+ if self.context is not None and len(self.context) > 0:
+ query_obj['context'] = self.context
+ if self.params is not None and len(self.params) > 0:
+ query_obj['parameters'] = self.params
+ if self.header:
+ query_obj['header'] = True
+ if self.result_format is not None:
+ query_obj['resultFormat'] = self.format
+ if self.sqlTypes:
+ query_obj['sqlTypesHeader'] = self.sqlTypes
+ if self.types:
+ query_obj['typesHeader'] = self.types
+ return query_obj
+
+ def result_format(self):
+ return self.format.lower()
+
+ def run(self):
+ return self.query_client.sql_query(self)
+
+def parse_rows(fmt, context, results):
+ if fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ rows = results['results']
+ elif fmt == consts.SQL_ARRAY:
+ rows = results
+ else:
+ return results
+ if not context.get('headers', False):
+ return rows
+ header_size = 1
+ if context.get('sqlTypesHeader', False):
+ header_size += 1
+ if context.get('typesHeader', False):
+ header_size += 1
+ return rows[header_size:]
+
+def label_non_null_cols(results):
+ if results is None or len(results) == 0:
+ return []
+ is_null = {}
+ for key in results[0].keys():
+ is_null[key] = True
+ for row in results:
+ for key, value in row.items():
+ if type(value) == str:
+ if value != '':
+ is_null[key] = False
+ elif type(value) == float:
+ if value != 0.0:
+ is_null[key] = False
+ elif value is not None:
+ is_null[key] = False
+ return is_null
+
+def filter_null_cols(results):
+ '''
+ Filter columns from a Druid result set by removing all null-like
+ columns. A column is considered null if all values for that column
+ are null. A value is null if it is either a JSON null, an empty
+ string, or a numeric 0. All rows are preserved, as is the order
+ of the remaining columns.
+ '''
+ if results is None or len(results) == 0:
+ return results
+ is_null = label_non_null_cols(results)
+ revised = []
+ for row in results:
+ new_row = {}
+ for key, value in row.items():
+ if is_null[key]:
+ continue
+ new_row[key] = value
+ revised.append(new_row)
+ return revised
+
+def parse_object_schema(results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ row = results[0]
+ for k, v in row.items():
+ druid_type = None
+ sql_type = None
+ if type(v) is str:
+ druid_type = consts.DRUID_STRING_TYPE
+ sql_type = consts.SQL_VARCHAR_TYPE
+ elif type(v) is int or type(v) is float:
+ druid_type = consts.DRUID_LONG_TYPE
+ sql_type = consts.SQL_BIGINT_TYPE
+ schema.append(ColumnSchema(k, sql_type, druid_type))
+ return schema
+
+def parse_array_schema(context, results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ has_headers = context.get(consts.HEADERS_KEY, False)
+ if not has_headers:
+ return schema
+ has_sql_types = context.get(consts.SQL_TYPES_HEADERS_KEY, False)
+ has_druid_types = context.get(consts.DRUID_TYPE_HEADERS_KEY, False)
+ size = len(results[0])
+ for i in range(size):
+ druid_type = None
+ if has_druid_types:
+ druid_type = results[1][i]
+ sql_type = None
+ if has_sql_types:
+ sql_type = results[2][i]
+ schema.append(ColumnSchema(results[0][i], sql_type, druid_type))
+ return schema
+
+def parse_schema(fmt, context, results):
+ if fmt == consts.SQL_OBJECT:
+ return parse_object_schema(results)
+ elif fmt == consts.SQL_ARRAY or fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ return parse_array_schema(context, results)
+ else:
+ return []
+
+def is_response_ok(http_response):
+ code = http_response.status_code
+ return code == requests.codes.ok or code == requests.codes.accepted
+
+class ColumnSchema:
+
+ def __init__(self, name, sql_type, druid_type):
+ self.name = name
+ self.sql_type = sql_type
+ self.druid_type = druid_type
+
+ def __str__(self):
+ return "{{name={}, SQL type={}, Druid type={}}}".format(self.name, self.sql_type, self.druid_type)
+
+class SqlQueryResult:
+ """
+ Defines the core protocol for Druid SQL queries.
+ """
+
+ def __init__(self, request, response):
+ self.http_response = response
+ self._json = None
+ self._rows = None
+ self._schema = None
+ self.request = request
+ self._error = None
+ self._id = None
+ if not is_response_ok(response):
+ try:
+ self._error = response.json()
+ except Exception:
+ self._error = response.text
+ if self._error is None or len(self._error) == 0:
+ self._error = "Failed with HTTP status {}".format(response.status_code)
+ try:
+ self._id = self.http_response.headers['X-Druid-SQL-Query-Id']
+ except KeyError:
+ self._error = "Query returned no query ID"
+
+ def result_format(self):
+ return self.request.result_format()
+
+ def ok(self):
+ """
+ Reports if the query succeeded.
Review Comment:
`ok()` for SqlQueryResult and QueryTaskResult have the same description. Should they be different? Or does it mean that the query rows and schema are available only if `ok()` returns true for both?
##########
examples/quickstart/jupyter-notebooks/druidapi/status.py:
##########
@@ -0,0 +1,99 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time
+
+STATUS_BASE = "/status"
+REQ_STATUS = STATUS_BASE
+REQ_HEALTH = STATUS_BASE + "/health"
+REQ_PROPERTIES = STATUS_BASE + "/properties"
+REQ_IN_CLUSTER = STATUS_BASE + "/selfDiscovered/status"
+
+ROUTER_BASE = '/druid/router/v1'
+REQ_BROKERS = ROUTER_BASE + '/brokers'
+
+class StatusClient:
+ '''
+ Client for status APIs. These APIs are available on all nodes.
+ If used with the router, they report the status of just the router.
+ '''
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ #-------- Common --------
+
+ def status(self):
+ """
+ Returns the Druid version, loaded extensions, memory used, total memory
+ and other useful information about the process.
+
+ GET `/status`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#process-information
+ """
+ return self.client.get_json(REQ_STATUS)
+
+ def is_healthy(self) -> bool:
+ """
+ Returns `True` if the node is healthy, an exception otherwise.
Review Comment:
Doesn't it return False otherwise? (instead of raising an exception)
##########
examples/quickstart/jupyter-notebooks/druidapi/error.py:
##########
@@ -0,0 +1,32 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+class ClientError(Exception):
+ """
+ Indicates an error with usage of the API.
Review Comment:
Should this refer to "the API client" to differentiate from an error arising from the Druid API?
##########
examples/quickstart/jupyter-notebooks/druidapi/display.py:
##########
@@ -0,0 +1,84 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+TEXT_TABLE = 0
+HTML_TABLE = 1
+
+class Display:
+
+ def __init__(self):
+ self.format = TEXT_TABLE
+ self.html_initialized = False
+
+ def text(self):
+ self.format = TEXT_TABLE
+
+ def html(self):
+ self.format = HTML_TABLE
+ if not self.html_initialized:
+ from .html_table import styles
+ styles()
+ self.html_initialized = True
+
+ def table(self):
+ if self.format == HTML_TABLE:
+ from .html_table import HtmlTable
+ return HtmlTable()
+ else:
+ from .text_table import TextTable
+ return TextTable()
+
+ def show_object_list(self, objects, cols):
+ list_to_table(self.table(), objects, cols)
+
+ def show_object(self, obj, labels):
+ object_to_table(self.table(), obj, labels)
+
+ def show_error(self, msg):
+ from .html_table import html_error
+ html_error("<b>ERROR: " + msg + "</b")
Review Comment:
Missing closing bracket?
##########
examples/quickstart/jupyter-notebooks/druidapi/druid.py:
##########
@@ -0,0 +1,70 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .rest import DruidRestClient
+from .status import StatusClient
+from .catalog import CatalogClient
+from .sql import QueryClient
+from .tasks import TaskClient
+from .datasource import DatasourceClient
+
+class DruidClient:
+
+ def __init__(self, router_endpoint):
+ self.rest_client = DruidRestClient(router_endpoint)
+ self.status_client = None
+ self.catalog_client = None
+ self.sql_client = None
+ self.tasks_client = None
+ self.datasource_client = None
+
+ def rest(self):
+ return self.rest_client
+
+ def trace(self, enable=True):
+ self.rest_client.enable_trace(enable)
+
+ def status(self, endpoint=None) -> StatusClient:
+ '''
+ Returns the status client for the router by default, else the status
+ endpoint for the specified endpoint.
+ '''
Review Comment:
```suggestion
"""
Returns the status client for the Router by default. Otherwise,
if an endpoint is specified, returns the status client
for the specified endpoint.
"""
```
##########
examples/quickstart/jupyter-notebooks/druidapi/datasource.py:
##########
@@ -0,0 +1,111 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests, time
+from .consts import COORD_BASE
+from .rest import check_error
+from .util import dict_get
+
+REQ_DATASOURCES = COORD_BASE + '/datasources'
+REQ_DATASOURCE = REQ_DATASOURCES + '/{}'
+
+# Segment load status
+REQ_DATASOURCES = COORD_BASE + '/datasources'
+REQ_DS_LOAD_STATUS = REQ_DATASOURCES + '/{}/loadstatus'
+
+class DatasourceClient:
+ '''
+ Client for status APIs. These APIs are available on all nodes.
+ If used with the router, they report the status of just the router.
+ '''
+
+ def __init__(self, rest_client):
+ self.rest_client = rest_client
+
+ def names(self, include_unused=False, include_disabled=False):
+ """
+ Returns a list of the names of data sources in the metadata store.
+
+ Parameters
+ ----------
+ include_unused : bool, default = False
+ if False, returns only datasources with at least one used segment
Review Comment:
```suggestion
If False, returns only datasources with at least one used segment.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/datasource.py:
##########
@@ -0,0 +1,111 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests, time
+from .consts import COORD_BASE
+from .rest import check_error
+from .util import dict_get
+
+REQ_DATASOURCES = COORD_BASE + '/datasources'
+REQ_DATASOURCE = REQ_DATASOURCES + '/{}'
+
+# Segment load status
+REQ_DATASOURCES = COORD_BASE + '/datasources'
+REQ_DS_LOAD_STATUS = REQ_DATASOURCES + '/{}/loadstatus'
+
+class DatasourceClient:
+ '''
+ Client for status APIs. These APIs are available on all nodes.
+ If used with the router, they report the status of just the router.
+ '''
+
+ def __init__(self, rest_client):
+ self.rest_client = rest_client
+
+ def names(self, include_unused=False, include_disabled=False):
+ """
+ Returns a list of the names of data sources in the metadata store.
+
+ Parameters
+ ----------
+ include_unused : bool, default = False
+ if False, returns only datasources with at least one used segment
+ in the cluster.
+
+ include_unused : bool, default = False
+ if False, returns only enamed datasources.
+
+ Reference
+ ---------
+ * `GET /druid/coordinator/v1/metadata/datasources`
+ * `GET /druid/coordinator/v1/metadata/datasources?includeUnused`
+ * `GET /druid/coordinator/v1/metadata/datasources?includeDisabled`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-4
Review Comment:
This link doesn't work for me. Should it go to https://druid.apache.org/docs/latest/operations/api-reference.html#metadata-store-information?
##########
examples/quickstart/jupyter-notebooks/druidapi/sql.py:
##########
@@ -0,0 +1,690 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time, requests
+from . import consts, display
+from .consts import ROUTER_BASE
+from .util import is_blank, dict_get
+from .error import DruidError, ClientError
+
+REQ_ROUTER_QUERY = ROUTER_BASE
+REQ_ROUTER_SQL = ROUTER_BASE + '/sql'
+REQ_ROUTER_SQL_TASK = REQ_ROUTER_SQL + '/task'
+
+class SqlRequest:
+
+ def __init__(self, query_client, sql):
+ self.query_client = query_client
+ self.sql = sql
+ self.context = None
+ self.params = None
+ self.header = False
+ self.format = consts.SQL_OBJECT
+ self.headers = None
+ self.types = None
+ self.sqlTypes = None
+
+ def with_format(self, result_format):
+ self.format = result_format
+ return self
+
+ def with_headers(self, sqlTypes=False, druidTypes=False):
+ self.headers = True
+ self.types = druidTypes
+ self.sqlTypes = sqlTypes
+ return self
+
+ def with_context(self, context):
+ if self.context is None:
+ self.context = context
+ else:
+ self.context.update(context)
+ return self
+
+ def with_parameters(self, params):
+ '''
+ Set the array of parameters. Parameters must each be a map of 'type'/'value' pairs:
+ {'type': the_type, 'value': the_value}. The type must be a valid SQL type
+ (in upper case). See the consts module for a list.
+ '''
+ if self.params is None:
+ self.params = params
+ else:
+ self.params.update(params)
+ return self
+
+ def add_parameter(self, value):
+ '''
+ Add one parameter value. Infers the type of the parameter from the Python type.
+ '''
+ if value is None:
+ raise ClientError("Druid does not support null parameter values")
+ data_type = None
+ value_type = type(value)
+ if value_type is str:
+ data_type = consts.SQL_VARCHAR_TYPE
+ elif value_type is int:
+ data_type = consts.SQL_BIGINT_TYPE
+ elif value_type is float:
+ data_type = consts.SQL_DOUBLE_TYPE
+ elif value_type is list:
+ data_type = consts.SQL_ARRAY_TYPE
+ else:
+ raise ClientError("Unsupported value type")
+ if self.params is None:
+ self.params = []
+ self.params.append({'type': data_type, 'value': value})
+
+ def response_header(self):
+ self.header = True
+ return self
+
+ def request_headers(self, headers):
+ self.headers = headers
+ return self
+
+ def to_request(self):
+ query_obj = {"query": self.sql}
+ if self.context is not None and len(self.context) > 0:
+ query_obj['context'] = self.context
+ if self.params is not None and len(self.params) > 0:
+ query_obj['parameters'] = self.params
+ if self.header:
+ query_obj['header'] = True
+ if self.result_format is not None:
+ query_obj['resultFormat'] = self.format
+ if self.sqlTypes:
+ query_obj['sqlTypesHeader'] = self.sqlTypes
+ if self.types:
+ query_obj['typesHeader'] = self.types
+ return query_obj
+
+ def result_format(self):
+ return self.format.lower()
+
+ def run(self):
+ return self.query_client.sql_query(self)
+
+def parse_rows(fmt, context, results):
+ if fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ rows = results['results']
+ elif fmt == consts.SQL_ARRAY:
+ rows = results
+ else:
+ return results
+ if not context.get('headers', False):
+ return rows
+ header_size = 1
+ if context.get('sqlTypesHeader', False):
+ header_size += 1
+ if context.get('typesHeader', False):
+ header_size += 1
+ return rows[header_size:]
+
+def label_non_null_cols(results):
+ if results is None or len(results) == 0:
+ return []
+ is_null = {}
+ for key in results[0].keys():
+ is_null[key] = True
+ for row in results:
+ for key, value in row.items():
+ if type(value) == str:
+ if value != '':
+ is_null[key] = False
+ elif type(value) == float:
+ if value != 0.0:
+ is_null[key] = False
+ elif value is not None:
+ is_null[key] = False
+ return is_null
+
+def filter_null_cols(results):
+ '''
+ Filter columns from a Druid result set by removing all null-like
+ columns. A column is considered null if all values for that column
+ are null. A value is null if it is either a JSON null, an empty
+ string, or a numeric 0. All rows are preserved, as is the order
+ of the remaining columns.
+ '''
+ if results is None or len(results) == 0:
+ return results
+ is_null = label_non_null_cols(results)
+ revised = []
+ for row in results:
+ new_row = {}
+ for key, value in row.items():
+ if is_null[key]:
+ continue
+ new_row[key] = value
+ revised.append(new_row)
+ return revised
+
+def parse_object_schema(results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ row = results[0]
+ for k, v in row.items():
+ druid_type = None
+ sql_type = None
+ if type(v) is str:
+ druid_type = consts.DRUID_STRING_TYPE
+ sql_type = consts.SQL_VARCHAR_TYPE
+ elif type(v) is int or type(v) is float:
+ druid_type = consts.DRUID_LONG_TYPE
+ sql_type = consts.SQL_BIGINT_TYPE
+ schema.append(ColumnSchema(k, sql_type, druid_type))
+ return schema
+
+def parse_array_schema(context, results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ has_headers = context.get(consts.HEADERS_KEY, False)
+ if not has_headers:
+ return schema
+ has_sql_types = context.get(consts.SQL_TYPES_HEADERS_KEY, False)
+ has_druid_types = context.get(consts.DRUID_TYPE_HEADERS_KEY, False)
+ size = len(results[0])
+ for i in range(size):
+ druid_type = None
+ if has_druid_types:
+ druid_type = results[1][i]
+ sql_type = None
+ if has_sql_types:
+ sql_type = results[2][i]
+ schema.append(ColumnSchema(results[0][i], sql_type, druid_type))
+ return schema
+
+def parse_schema(fmt, context, results):
+ if fmt == consts.SQL_OBJECT:
+ return parse_object_schema(results)
+ elif fmt == consts.SQL_ARRAY or fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ return parse_array_schema(context, results)
+ else:
+ return []
+
+def is_response_ok(http_response):
+ code = http_response.status_code
+ return code == requests.codes.ok or code == requests.codes.accepted
+
+class ColumnSchema:
+
+ def __init__(self, name, sql_type, druid_type):
+ self.name = name
+ self.sql_type = sql_type
+ self.druid_type = druid_type
+
+ def __str__(self):
+ return "{{name={}, SQL type={}, Druid type={}}}".format(self.name, self.sql_type, self.druid_type)
+
+class SqlQueryResult:
+ """
+ Defines the core protocol for Druid SQL queries.
+ """
+
+ def __init__(self, request, response):
+ self.http_response = response
+ self._json = None
+ self._rows = None
+ self._schema = None
+ self.request = request
+ self._error = None
+ self._id = None
+ if not is_response_ok(response):
+ try:
+ self._error = response.json()
+ except Exception:
+ self._error = response.text
+ if self._error is None or len(self._error) == 0:
+ self._error = "Failed with HTTP status {}".format(response.status_code)
+ try:
+ self._id = self.http_response.headers['X-Druid-SQL-Query-Id']
+ except KeyError:
+ self._error = "Query returned no query ID"
+
+ def result_format(self):
+ return self.request.result_format()
+
+ def ok(self):
+ """
+ Reports if the query succeeded.
+
+ The query rows and schema are available only if ok() returns True.
+ """
+ return is_response_ok(self.http_response)
+
+ def error_msg(self):
+ err = self.error()
+ if err is None:
+ return "unknown"
+ if type(err) is str:
+ return err
+ msg = err.get("error")
+ text = err.get("errorMessage")
+ if msg is None and text is None:
+ return "unknown"
+ if msg is None:
+ return text
+ if text is None:
+ return msg
+ return msg + ": " + text
+
+ def id(self):
+ """
+ Returns the unique identifier for the query.
+ """
+ return self._id
+
+ def non_null(self):
+ if not self.ok():
+ return None
+ if self.result_format() != consts.SQL_OBJECT:
+ return None
+ return filter_null_cols(self.rows())
+
+ def as_array(self):
+ if self.result_format() == consts.SQL_OBJECT:
+ rows = []
+ for obj in self.rows():
+ rows.append([v for v in obj.values()])
+ return rows
+ else:
+ return self.rows()
+
+ def error(self):
+ """
+ If the query fails, returns the error, if any provided by Druid.
+ """
+ if self.ok():
+ return None
+ if self._error is not None:
+ return self._error
+ if self.http_response is None:
+ return { "error": "unknown"}
+ if is_response_ok(self.http_response):
+ return None
+ return {"error": "HTTP {}".format(self.http_response.status_code)}
+
+ def json(self):
+ if not self.ok():
+ return None
+ if self._json is None:
+ self._json = self.http_response.json()
+ return self._json
+
+ def rows(self):
+ """
+ Returns the rows of data for the query.
+
+ Druid supports many data formats. The method makes its best
+ attempt to map the format into an array of rows of some sort.
+ """
+ if self._rows is None:
+ json = self.json()
+ if json is None:
+ return self.http_response.text
+ self._rows = parse_rows(self.result_format(), self.request.context, json)
+ return self._rows
+
+ def schema(self):
+ """
+ Returns the data schema as a list of ColumnSchema objects.
+
+ Druid supports many data formats, not all of them provide
+ schema information. This method makes its best attempt to
Review Comment:
```suggestion
Druid supports many data formats; not all of them provide
schema information. This method makes a best effort attempt to
```
##########
examples/quickstart/jupyter-notebooks/druidapi/sql.py:
##########
@@ -0,0 +1,690 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time, requests
+from . import consts, display
+from .consts import ROUTER_BASE
+from .util import is_blank, dict_get
+from .error import DruidError, ClientError
+
+REQ_ROUTER_QUERY = ROUTER_BASE
+REQ_ROUTER_SQL = ROUTER_BASE + '/sql'
+REQ_ROUTER_SQL_TASK = REQ_ROUTER_SQL + '/task'
+
+class SqlRequest:
+
+ def __init__(self, query_client, sql):
+ self.query_client = query_client
+ self.sql = sql
+ self.context = None
+ self.params = None
+ self.header = False
+ self.format = consts.SQL_OBJECT
+ self.headers = None
+ self.types = None
+ self.sqlTypes = None
+
+ def with_format(self, result_format):
+ self.format = result_format
+ return self
+
+ def with_headers(self, sqlTypes=False, druidTypes=False):
+ self.headers = True
+ self.types = druidTypes
+ self.sqlTypes = sqlTypes
+ return self
+
+ def with_context(self, context):
+ if self.context is None:
+ self.context = context
+ else:
+ self.context.update(context)
+ return self
+
+ def with_parameters(self, params):
+ '''
+ Set the array of parameters. Parameters must each be a map of 'type'/'value' pairs:
+ {'type': the_type, 'value': the_value}. The type must be a valid SQL type
+ (in upper case). See the consts module for a list.
+ '''
+ if self.params is None:
+ self.params = params
+ else:
+ self.params.update(params)
+ return self
+
+ def add_parameter(self, value):
+ '''
+ Add one parameter value. Infers the type of the parameter from the Python type.
+ '''
+ if value is None:
+ raise ClientError("Druid does not support null parameter values")
+ data_type = None
+ value_type = type(value)
+ if value_type is str:
+ data_type = consts.SQL_VARCHAR_TYPE
+ elif value_type is int:
+ data_type = consts.SQL_BIGINT_TYPE
+ elif value_type is float:
+ data_type = consts.SQL_DOUBLE_TYPE
+ elif value_type is list:
+ data_type = consts.SQL_ARRAY_TYPE
+ else:
+ raise ClientError("Unsupported value type")
+ if self.params is None:
+ self.params = []
+ self.params.append({'type': data_type, 'value': value})
+
+ def response_header(self):
+ self.header = True
+ return self
+
+ def request_headers(self, headers):
+ self.headers = headers
+ return self
+
+ def to_request(self):
+ query_obj = {"query": self.sql}
+ if self.context is not None and len(self.context) > 0:
+ query_obj['context'] = self.context
+ if self.params is not None and len(self.params) > 0:
+ query_obj['parameters'] = self.params
+ if self.header:
+ query_obj['header'] = True
+ if self.result_format is not None:
+ query_obj['resultFormat'] = self.format
+ if self.sqlTypes:
+ query_obj['sqlTypesHeader'] = self.sqlTypes
+ if self.types:
+ query_obj['typesHeader'] = self.types
+ return query_obj
+
+ def result_format(self):
+ return self.format.lower()
+
+ def run(self):
+ return self.query_client.sql_query(self)
+
+def parse_rows(fmt, context, results):
+ if fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ rows = results['results']
+ elif fmt == consts.SQL_ARRAY:
+ rows = results
+ else:
+ return results
+ if not context.get('headers', False):
+ return rows
+ header_size = 1
+ if context.get('sqlTypesHeader', False):
+ header_size += 1
+ if context.get('typesHeader', False):
+ header_size += 1
+ return rows[header_size:]
+
+def label_non_null_cols(results):
+ if results is None or len(results) == 0:
+ return []
+ is_null = {}
+ for key in results[0].keys():
+ is_null[key] = True
+ for row in results:
+ for key, value in row.items():
+ if type(value) == str:
+ if value != '':
+ is_null[key] = False
+ elif type(value) == float:
+ if value != 0.0:
+ is_null[key] = False
+ elif value is not None:
+ is_null[key] = False
+ return is_null
+
+def filter_null_cols(results):
+ '''
+ Filter columns from a Druid result set by removing all null-like
+ columns. A column is considered null if all values for that column
+ are null. A value is null if it is either a JSON null, an empty
+ string, or a numeric 0. All rows are preserved, as is the order
+ of the remaining columns.
+ '''
+ if results is None or len(results) == 0:
+ return results
+ is_null = label_non_null_cols(results)
+ revised = []
+ for row in results:
+ new_row = {}
+ for key, value in row.items():
+ if is_null[key]:
+ continue
+ new_row[key] = value
+ revised.append(new_row)
+ return revised
+
+def parse_object_schema(results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ row = results[0]
+ for k, v in row.items():
+ druid_type = None
+ sql_type = None
+ if type(v) is str:
+ druid_type = consts.DRUID_STRING_TYPE
+ sql_type = consts.SQL_VARCHAR_TYPE
+ elif type(v) is int or type(v) is float:
+ druid_type = consts.DRUID_LONG_TYPE
+ sql_type = consts.SQL_BIGINT_TYPE
+ schema.append(ColumnSchema(k, sql_type, druid_type))
+ return schema
+
+def parse_array_schema(context, results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ has_headers = context.get(consts.HEADERS_KEY, False)
+ if not has_headers:
+ return schema
+ has_sql_types = context.get(consts.SQL_TYPES_HEADERS_KEY, False)
+ has_druid_types = context.get(consts.DRUID_TYPE_HEADERS_KEY, False)
+ size = len(results[0])
+ for i in range(size):
+ druid_type = None
+ if has_druid_types:
+ druid_type = results[1][i]
+ sql_type = None
+ if has_sql_types:
+ sql_type = results[2][i]
+ schema.append(ColumnSchema(results[0][i], sql_type, druid_type))
+ return schema
+
+def parse_schema(fmt, context, results):
+ if fmt == consts.SQL_OBJECT:
+ return parse_object_schema(results)
+ elif fmt == consts.SQL_ARRAY or fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ return parse_array_schema(context, results)
+ else:
+ return []
+
+def is_response_ok(http_response):
+ code = http_response.status_code
+ return code == requests.codes.ok or code == requests.codes.accepted
+
+class ColumnSchema:
+
+ def __init__(self, name, sql_type, druid_type):
+ self.name = name
+ self.sql_type = sql_type
+ self.druid_type = druid_type
+
+ def __str__(self):
+ return "{{name={}, SQL type={}, Druid type={}}}".format(self.name, self.sql_type, self.druid_type)
+
+class SqlQueryResult:
+ """
+ Defines the core protocol for Druid SQL queries.
+ """
+
+ def __init__(self, request, response):
+ self.http_response = response
+ self._json = None
+ self._rows = None
+ self._schema = None
+ self.request = request
+ self._error = None
+ self._id = None
+ if not is_response_ok(response):
+ try:
+ self._error = response.json()
+ except Exception:
+ self._error = response.text
+ if self._error is None or len(self._error) == 0:
+ self._error = "Failed with HTTP status {}".format(response.status_code)
+ try:
+ self._id = self.http_response.headers['X-Druid-SQL-Query-Id']
+ except KeyError:
+ self._error = "Query returned no query ID"
+
+ def result_format(self):
+ return self.request.result_format()
+
+ def ok(self):
+ """
+ Reports if the query succeeded.
+
+ The query rows and schema are available only if ok() returns True.
+ """
+ return is_response_ok(self.http_response)
+
+ def error_msg(self):
+ err = self.error()
+ if err is None:
+ return "unknown"
+ if type(err) is str:
+ return err
+ msg = err.get("error")
+ text = err.get("errorMessage")
+ if msg is None and text is None:
+ return "unknown"
+ if msg is None:
+ return text
+ if text is None:
+ return msg
+ return msg + ": " + text
+
+ def id(self):
+ """
+ Returns the unique identifier for the query.
+ """
+ return self._id
+
+ def non_null(self):
+ if not self.ok():
+ return None
+ if self.result_format() != consts.SQL_OBJECT:
+ return None
+ return filter_null_cols(self.rows())
+
+ def as_array(self):
+ if self.result_format() == consts.SQL_OBJECT:
+ rows = []
+ for obj in self.rows():
+ rows.append([v for v in obj.values()])
+ return rows
+ else:
+ return self.rows()
+
+ def error(self):
+ """
+ If the query fails, returns the error, if any provided by Druid.
+ """
+ if self.ok():
+ return None
+ if self._error is not None:
+ return self._error
+ if self.http_response is None:
+ return { "error": "unknown"}
+ if is_response_ok(self.http_response):
+ return None
+ return {"error": "HTTP {}".format(self.http_response.status_code)}
+
+ def json(self):
+ if not self.ok():
+ return None
+ if self._json is None:
+ self._json = self.http_response.json()
+ return self._json
+
+ def rows(self):
+ """
+ Returns the rows of data for the query.
+
+ Druid supports many data formats. The method makes its best
+ attempt to map the format into an array of rows of some sort.
+ """
+ if self._rows is None:
+ json = self.json()
+ if json is None:
+ return self.http_response.text
+ self._rows = parse_rows(self.result_format(), self.request.context, json)
+ return self._rows
+
+ def schema(self):
+ """
+ Returns the data schema as a list of ColumnSchema objects.
+
+ Druid supports many data formats, not all of them provide
+ schema information. This method makes its best attempt to
+ extract the schema from the query results.
+ """
+ if self._schema is None:
+ self._schema = parse_schema(self.result_format(), self.request.context, self.json())
+ return self._schema
+
+ def show(self, non_null=False):
+ data = None
+ if non_null:
+ data = self.non_null()
+ if data is None:
+ data = self.as_array()
+ if data is None or len(data) == 0:
+ display.display.show_message("Query returned no results")
+ return
+ disp = display.display.table()
+ disp.headers([c.name for c in self.schema()])
+ disp.show(data)
+
+ def show_schema(self):
+ disp = display.display.table()
+ disp.headers(['Name', 'SQL Type', 'Druid Type'])
+ data = []
+ for c in self.schema():
+ data.append([c.name, c.sql_type, c.druid_type])
+ disp.show(data)
+
+class QueryTaskResult:
+
+ def __init__(self, request, response):
+ self._request = request
+ self.http_response = response
+ self._status = None
+ self._results = None
+ self._details = None
+ self._schema = None
+ self._rows = None
+ self._reports = None
+ self._schema = None
+ self._results = None
+ self._error = None
+ self._id = None
+ if not is_response_ok(response):
+ self._state = consts.FAILED_STATE
+ try:
+ self._error = response.json()
+ except Exception:
+ self._error = response.text
+ if self._error is None or len(self._error) == 0:
+ self._error = "Failed with HTTP status {}".format(response.status_code)
+ return
+
+ # Typical response:
+ # {'taskId': '6f7b514a446d4edc9d26a24d4bd03ade_fd8e242b-7d93-431d-b65b-2a512116924c_bjdlojgj',
+ # 'state': 'RUNNING'}
+ self.response_obj = response.json()
+ self._id = self.response_obj['taskId']
+ self._state = self.response_obj['state']
+
+ def ok(self):
+ """
+ Reports if the query succeeded.
+
+ The query rows and schema are available only if ok() returns True.
+ """
+ return self._error is None
+
+ def id(self):
+ return self._id
+
+ def _tasks(self):
+ return self._request.query_client.druid_client.tasks()
+
+ def status(self):
+ """
+ Polls Druid for an update on the query run status.
+ """
+ self.check_valid()
+ # Example:
+ # {'task': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'status': {
+ # 'id': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'groupId': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'type': 'talaria0', 'createdTime': '2022-04-28T23:19:50.331Z',
+ # 'queueInsertionTime': '1970-01-01T00:00:00.000Z',
+ # 'statusCode': 'RUNNING', 'status': 'RUNNING', 'runnerStatusCode': 'PENDING',
+ # 'duration': -1, 'location': {'host': None, 'port': -1, 'tlsPort': -1},
+ # 'dataSource': 'w000', 'errorMsg': None}}
+ self._status = self._tasks().task_status(self._id)
+ self._state = self._status['status']['status']
+ if self._state == consts.FAILED_STATE:
+ self._error = self._status['status']['errorMsg']
+ return self._status
+
+ def done(self):
+ """
+ Reports if the query is done: succeeded or failed.
+ """
+ return self._state == consts.FAILED_STATE or self._state == consts.SUCCESS_STATE
+
+ def succeeded(self):
+ """
+ Reports if the query succeeded.
+ """
+ return self._state == consts.SUCCESS_STATE
+
+ def state(self):
+ """
+ Reports the engine-specific query state.
+
+ Updated after each call to status().
+ """
+ return self._state
+
+ def error(self):
+ return self._error
+
+ def error_msg(self):
+ err = self.error()
+ if err is None:
+ return "unknown"
+ if type(err) is str:
+ return err
+ msg = dict_get(err, "error")
+ text = dict_get(err, "errorMessage")
+ if msg is None and text is None:
+ return "unknown"
+ if text is not None:
+ text = text.replace('\\n', '\n')
+ if msg is None:
+ return text
+ if text is None:
+ return msg
+ return msg + ": " + text
+
+ def join(self):
+ if not self.done():
+ self.status()
+ while not self.done():
+ time.sleep(0.5)
+ self.status()
+ return self.succeeded()
+
+ def check_valid(self):
+ if self._id is None:
+ raise ClientError("Operation is invalid on a failed query")
+
+ def wait_done(self):
+ if not self.join():
+ raise DruidError("Query failed: " + self.error_msg())
+
+ def wait(self):
+ self.wait_done()
+ return self.rows()
+
+ def reports(self) -> dict:
+ self.check_valid()
+ if self._reports is None:
+ self.join()
+ self._reports = self._tasks().task_reports(self._id)
+ return self._reports
+
+ def results(self):
+ if self._results is None:
+ rpts = self.reports()
+ self._results = rpts['multiStageQuery']['payload']['results']
+ return self._results
+
+ def schema(self):
+ if self._schema is None:
+ results = self.results()
+ sig = results['signature']
+ sqlTypes = results['sqlTypeNames']
+ size = len(sig)
+ self._schema = []
+ for i in range(size):
+ self._schema.append(ColumnSchema(sig[i]['name'], sqlTypes[i], sig[i]['type']))
+ return self._schema
+
+ def rows(self):
+ if self._rows is None:
+ results = self.results()
+ self._rows = results['results']
+ return self._rows
+
+ def show(self, non_null=False):
+ data = self.rows()
+ if non_null:
+ data = filter_null_cols(data)
+ disp = display.display.table()
+ disp.headers([c.name for c in self.schema()])
+ disp.show(data)
+
+class QueryClient:
+
+ def __init__(self, druid, rest_client=None):
+ self.druid_client = druid
+ self._rest_client = druid.rest_client if rest_client is None else rest_client
+
+ def rest_client(self):
+ return self._rest_client
+
+ def _prepare_query(self, request):
+ if request is None:
+ raise ClientError("No query provided.")
+ if type(request) == str:
+ request = self.sql_request(request)
+ if is_blank(request.sql):
+ raise ClientError("No query provided.")
+ if self.rest_client().trace:
+ print(request.sql)
+ query_obj = request.to_request()
+ return (request, query_obj)
+
+ def sql_query(self, request) -> SqlQueryResult:
+ '''
+ Submit a SQL query with control over the context, parameters and other
+ options. Returns a response with either a detailed error message, or
+ the rows and query ID.
+ '''
+ request, query_obj = self._prepare_query(request)
+ r = self.rest_client().post_only_json(REQ_ROUTER_SQL, query_obj, headers=request.headers)
+ return SqlQueryResult(request, r)
+
+ def sql(self, sql, *args):
+ if len(args) > 0:
+ sql = sql.result_format(*args)
+ resp = self.sql_query(sql)
+ if resp.ok():
+ return resp.rows()
+ raise ClientError(resp.error_msg())
+
+ def explain_sql(self, query):
+ """
+ Run an EXPLAIN PLAN FOR query for the given query.
+
+ Returns
+ -------
+ An object with the plan JSON parsed into Python objects:
+ plan: the query plan
+ columns: column schema
+ tables: dictionary of name/type pairs
+ """
+ if is_blank(query):
+ raise ClientError("No query provided.")
+ results = self.sql('EXPLAIN PLAN FOR ' + query)
+ return results[0]
+
+ def sql_request(self, sql):
+ return SqlRequest(self, sql)
+
+ def show(self, query):
+ result = self.sql_query(query)
+ if result.ok():
+ result.show()
+ else:
+ display.display.show_error(result.error_msg())
+
+ def task(self, request):
+ request, query_obj = self._prepare_query(request)
+ r = self.rest_client().post_only_json(REQ_ROUTER_SQL_TASK, query_obj, headers=request.headers)
+ return QueryTaskResult(request, r)
+
+ def run_task(self, request):
+ resp = self.task(request)
+ if not resp.ok():
+ raise ClientError(resp.error_msg())
+ resp.wait_done()
+
+ def _tables_query(self, schema):
+ return self.sql_query('''
+ SELECT TABLE_NAME AS TableName
+ FROM INFORMATION_SCHEMA.TABLES
+ WHERE TABLE_SCHEMA = '{}'
+ ORDER BY TABLE_NAME
+ '''.format(schema))
+
+ def tables(self, schema=consts.DRUID_SCHEMA):
+ return self._tables_query(schema).rows()
+
+ def show_tables(self, schema=consts.DRUID_SCHEMA):
+ self._tables_query(schema).show()
+
+ def _schemas_query(self):
+ return self.sql_query('''
+ SELECT SCHEMA_NAME AS SchemaName
+ FROM INFORMATION_SCHEMA.SCHEMATA
+ ORDER BY SCHEMA_NAME
+ ''')
+
+ def show_schemas(self):
+ self._schemas_query().show()
+
+ def describe_table(self, part1, part2=None):
Review Comment:
What is part1 and part2? Is this method only intended to be used internally? Same question on `describe_function()`
##########
examples/quickstart/jupyter-notebooks/druidapi/rest.py:
##########
@@ -0,0 +1,180 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests
+from .util import dict_get, is_blank
+from urllib.parse import quote
+from .error import ClientError
+
+def check_error(response):
+ """
+ Raises a requests HttpError if the response code is not OK or Accepted.
Review Comment:
```suggestion
Takes a response object from the requests library and checks for error.
Raises a requests HttpError if the response code is not OK or Accepted.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/error.py:
##########
@@ -0,0 +1,32 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+class ClientError(Exception):
+ """
+ Indicates an error with usage of the API.
+ """
+
+ def __init__(self, msg):
+ self.message = msg
+
+class DruidError(Exception):
+ """
+ Indicates that something went wrong on Druid: often as a result of a
Review Comment:
```suggestion
Indicates that something went wrong on Druid, often as a result of a
```
##########
examples/quickstart/jupyter-notebooks/druidapi/tasks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .consts import OVERLORD_BASE
+
+# Tasks
+REQ_TASKS = OVERLORD_BASE + '/tasks'
+REQ_POST_TASK = OVERLORD_BASE + '/task'
+REQ_GET_TASK = REQ_POST_TASK + '/{}'
+REQ_TASK_STATUS = REQ_GET_TASK + '/status'
+REQ_TASK_REPORTS = REQ_GET_TASK + '/reports'
+REQ_END_TASK = REQ_GET_TASK
+REQ_END_DS_TASKS = REQ_END_TASK + '/shutdownAllTasks'
+
+class TaskClient:
+ """
+ Client for task-related APIs. The APIs connect through the Router to
+ the Overlord.
+ """
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def tasks(self, state=None, table=None, type=None, max=None, created_time_interval=None):
+ '''
+ Retrieve list of tasks.
+
+ Parameters
+ ----------
+ state : str, default = None
+ Filter list of tasks by task state. Valid options are "running",
+ "complete", "waiting", and "pending". Constants are defined for
+ each of these in the `consts` file.
+ table : str, default = None
+ Return tasks filtered by Druid table (datasource).
+ created_time_interval : str, Default = None
+ Return tasks created within the specified interval.
+ max : int, default = None
+ Maximum number of "complete" tasks to return. Only applies when state is set to "complete".
+ type : str, default = None
+ filter tasks by task type.
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/tasks`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
Review Comment:
The "get-15" link doesn't work for me. Should it point to https://druid.apache.org/docs/latest/operations/api-reference.html#tasks?
##########
examples/quickstart/jupyter-notebooks/druidapi/tasks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .consts import OVERLORD_BASE
+
+# Tasks
+REQ_TASKS = OVERLORD_BASE + '/tasks'
+REQ_POST_TASK = OVERLORD_BASE + '/task'
+REQ_GET_TASK = REQ_POST_TASK + '/{}'
+REQ_TASK_STATUS = REQ_GET_TASK + '/status'
+REQ_TASK_REPORTS = REQ_GET_TASK + '/reports'
+REQ_END_TASK = REQ_GET_TASK
+REQ_END_DS_TASKS = REQ_END_TASK + '/shutdownAllTasks'
+
+class TaskClient:
+ """
+ Client for task-related APIs. The APIs connect through the Router to
+ the Overlord.
+ """
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def tasks(self, state=None, table=None, type=None, max=None, created_time_interval=None):
+ '''
+ Retrieve list of tasks.
+
+ Parameters
+ ----------
+ state : str, default = None
+ Filter list of tasks by task state. Valid options are "running",
+ "complete", "waiting", and "pending". Constants are defined for
+ each of these in the `consts` file.
+ table : str, default = None
+ Return tasks filtered by Druid table (datasource).
+ created_time_interval : str, Default = None
+ Return tasks created within the specified interval.
+ max : int, default = None
+ Maximum number of "complete" tasks to return. Only applies when state is set to "complete".
+ type : str, default = None
+ filter tasks by task type.
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/tasks`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ params = {}
+ if state is not None:
+ params['state'] = state
+ if table is not None:
+ params['datasource'] = table
+ if type is not None:
+ params['type'] = type
+ if max is not None:
+ params['max'] = max
+ if created_time_interval is not None:
+ params['createdTimeInterval'] = created_time_interval
+ return self.client.get_json(REQ_TASKS, params=params)
+
+ def task(self, task_id):
+ """
+ Retrieve the "payload" of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ """
+ return self.client.get_json(REQ_GET_TASK, args=[task_id])
+
+ def task_status(self, task_id):
+ '''
+ Retrieve the status of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}/status`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ return self.client.get_json(REQ_TASK_STATUS, args=[task_id])
+
+ def task_reports(self, task_id):
+ '''
+ Retrieve a task completion report for a task.
+ Only works for completed tasks.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}/reports`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ return self.client.get_json(REQ_TASK_REPORTS, args=[task_id])
+
+ def submit_task(self, payload):
+ """
+ Submit a task or supervisor specs to the Overlord.
+
+ Returns the taskId of the submitted task.
+
+ Parameters
+ ----------
+ payload : object
+ The task object. Serialized to JSON.
+
+ Reference
+ ---------
+ `POST /druid/indexer/v1/task`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#post-5
Review Comment:
This link doesn't seem valid for me.
##########
examples/quickstart/jupyter-notebooks/druidapi/rest.py:
##########
@@ -0,0 +1,180 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests
+from .util import dict_get, is_blank
+from urllib.parse import quote
+from .error import ClientError
+
+def check_error(response):
+ """
+ Raises a requests HttpError if the response code is not OK or Accepted.
+
+ If the response included a JSON payload, then the message is extracted
+ from that payload, else the message is from requests. The JSON
+ payload, if any, is returned in the json field of the error.
+ """
+ code = response.status_code
+ if code == requests.codes.ok or code == requests.codes.accepted:
+ return
+ error = None
+ json = None
+ try:
+ json = response.json()
+ except Exception:
+ # If we can't get the JSON, just move on, we'll figure
+ # things out another way.
+ pass
+ msg = dict_get(json, 'errorMessage')
+ if msg is None:
+ msg = dict_get(json, 'error')
+ if not is_blank(msg):
+ raise ClientError(msg)
+ if code == requests.codes.not_found and error is None:
+ error = "Not found"
+ if error is not None:
+ response.reason = error
+ try:
+ response.raise_for_status()
+ except Exception as e:
+ e.json = json
+ raise e
+
+class DruidRestClient:
+ '''
+ Wrapper around the basic Druid REST API operations using the
+ requests Python package. Handles the grunt work of building up
+ URLs, working with JSON, etc.
+ '''
+
+ def __init__(self, endpoint):
+ self.endpoint = endpoint
+ self.trace = False
+ self.session = requests.Session()
+
+ def enable_trace(self, flag=True):
+ self.trace = flag
+
+ def build_url(self, req, args=None) -> str:
+ """
+ Returns the full URL for a REST call given the relative request API and
+ optional parameters to fill placeholders within the request URL.
+
+ Parameters
+ ----------
+ req : str
+ relative URL, with optional {} placeholders
+
+ args : list
+ optional list of values to match {} placeholders
Review Comment:
```suggestion
Optional list of values to match {} placeholders
```
##########
examples/quickstart/jupyter-notebooks/druidapi/rest.py:
##########
@@ -0,0 +1,180 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests
+from .util import dict_get, is_blank
+from urllib.parse import quote
+from .error import ClientError
+
+def check_error(response):
+ """
+ Raises a requests HttpError if the response code is not OK or Accepted.
+
+ If the response included a JSON payload, then the message is extracted
+ from that payload, else the message is from requests. The JSON
+ payload, if any, is returned in the json field of the error.
Review Comment:
```suggestion
Extracts the error message from the JSON payload, if included in the
response, else takes the error message from the requests library.
Returns the JSON payload, if included, in the json field of the raised error.
```
Not sure if this is what you're trying to say
##########
examples/quickstart/jupyter-notebooks/druidapi/tasks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .consts import OVERLORD_BASE
+
+# Tasks
+REQ_TASKS = OVERLORD_BASE + '/tasks'
+REQ_POST_TASK = OVERLORD_BASE + '/task'
+REQ_GET_TASK = REQ_POST_TASK + '/{}'
+REQ_TASK_STATUS = REQ_GET_TASK + '/status'
+REQ_TASK_REPORTS = REQ_GET_TASK + '/reports'
+REQ_END_TASK = REQ_GET_TASK
+REQ_END_DS_TASKS = REQ_END_TASK + '/shutdownAllTasks'
+
+class TaskClient:
+ """
+ Client for task-related APIs. The APIs connect through the Router to
+ the Overlord.
+ """
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def tasks(self, state=None, table=None, type=None, max=None, created_time_interval=None):
+ '''
+ Retrieve list of tasks.
+
+ Parameters
+ ----------
+ state : str, default = None
+ Filter list of tasks by task state. Valid options are "running",
+ "complete", "waiting", and "pending". Constants are defined for
+ each of these in the `consts` file.
+ table : str, default = None
+ Return tasks filtered by Druid table (datasource).
+ created_time_interval : str, Default = None
+ Return tasks created within the specified interval.
+ max : int, default = None
+ Maximum number of "complete" tasks to return. Only applies when state is set to "complete".
+ type : str, default = None
+ filter tasks by task type.
Review Comment:
```suggestion
Filter list of tasks by task state. Valid options are "running",
"complete", "waiting", and "pending". Constants are defined for
each of these in the `consts` module.
table : str, default = None
Return tasks filtered by Druid table (datasource).
created_time_interval : str, default = None
Return tasks created within the specified interval.
max : int, default = None
Maximum number of "complete" tasks to return. Only applies when state is set to "complete".
type : str, default = None
Filter tasks by task type.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/tasks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .consts import OVERLORD_BASE
+
+# Tasks
+REQ_TASKS = OVERLORD_BASE + '/tasks'
+REQ_POST_TASK = OVERLORD_BASE + '/task'
+REQ_GET_TASK = REQ_POST_TASK + '/{}'
+REQ_TASK_STATUS = REQ_GET_TASK + '/status'
+REQ_TASK_REPORTS = REQ_GET_TASK + '/reports'
+REQ_END_TASK = REQ_GET_TASK
+REQ_END_DS_TASKS = REQ_END_TASK + '/shutdownAllTasks'
+
+class TaskClient:
+ """
+ Client for task-related APIs. The APIs connect through the Router to
+ the Overlord.
+ """
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def tasks(self, state=None, table=None, type=None, max=None, created_time_interval=None):
+ '''
+ Retrieve list of tasks.
Review Comment:
```suggestion
Retrieves list of tasks.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/rest.py:
##########
@@ -0,0 +1,180 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests
+from .util import dict_get, is_blank
+from urllib.parse import quote
+from .error import ClientError
+
+def check_error(response):
+ """
+ Raises a requests HttpError if the response code is not OK or Accepted.
+
+ If the response included a JSON payload, then the message is extracted
+ from that payload, else the message is from requests. The JSON
+ payload, if any, is returned in the json field of the error.
+ """
+ code = response.status_code
+ if code == requests.codes.ok or code == requests.codes.accepted:
+ return
+ error = None
+ json = None
+ try:
+ json = response.json()
+ except Exception:
+ # If we can't get the JSON, just move on, we'll figure
+ # things out another way.
+ pass
+ msg = dict_get(json, 'errorMessage')
+ if msg is None:
+ msg = dict_get(json, 'error')
+ if not is_blank(msg):
+ raise ClientError(msg)
+ if code == requests.codes.not_found and error is None:
+ error = "Not found"
+ if error is not None:
+ response.reason = error
+ try:
+ response.raise_for_status()
+ except Exception as e:
+ e.json = json
+ raise e
+
+class DruidRestClient:
+ '''
+ Wrapper around the basic Druid REST API operations using the
+ requests Python package. Handles the grunt work of building up
+ URLs, working with JSON, etc.
+ '''
+
+ def __init__(self, endpoint):
+ self.endpoint = endpoint
+ self.trace = False
+ self.session = requests.Session()
+
+ def enable_trace(self, flag=True):
+ self.trace = flag
+
+ def build_url(self, req, args=None) -> str:
+ """
+ Returns the full URL for a REST call given the relative request API and
+ optional parameters to fill placeholders within the request URL.
+
+ Parameters
+ ----------
+ req : str
+ relative URL, with optional {} placeholders
+
+ args : list
+ optional list of values to match {} placeholders
+ in the URL.
+ """
+ url = self.endpoint + req
+ if args is not None:
+ quoted = [quote(arg) for arg in args]
+ url = url.format(*quoted)
+ return url
+
+ def get(self, req, args=None, params=None, require_ok=True) -> requests.Request:
+ '''
+ Generic GET request to this service.
+
+ Parameters
+ ----------
+ req: str
+ The request URL without host, port or query string.
+ Example: `/status`
+
+ args: [str], default = None
+ Optional parameters to fill in to the URL.
+ Example: `/customer/{}`
+
+ params: dict, default = None
+ Optional map of query variables to send in
+ the URL. Query parameters are the name/values pairs
Review Comment:
```suggestion
the URL. Query parameters are the name/value pairs
```
##########
examples/quickstart/jupyter-notebooks/druidapi/rest.py:
##########
@@ -0,0 +1,180 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests
+from .util import dict_get, is_blank
+from urllib.parse import quote
+from .error import ClientError
+
+def check_error(response):
+ """
+ Raises a requests HttpError if the response code is not OK or Accepted.
+
+ If the response included a JSON payload, then the message is extracted
+ from that payload, else the message is from requests. The JSON
+ payload, if any, is returned in the json field of the error.
+ """
+ code = response.status_code
+ if code == requests.codes.ok or code == requests.codes.accepted:
+ return
+ error = None
+ json = None
+ try:
+ json = response.json()
+ except Exception:
+ # If we can't get the JSON, just move on, we'll figure
+ # things out another way.
+ pass
+ msg = dict_get(json, 'errorMessage')
+ if msg is None:
+ msg = dict_get(json, 'error')
+ if not is_blank(msg):
+ raise ClientError(msg)
+ if code == requests.codes.not_found and error is None:
+ error = "Not found"
+ if error is not None:
+ response.reason = error
+ try:
+ response.raise_for_status()
+ except Exception as e:
+ e.json = json
+ raise e
+
+class DruidRestClient:
+ '''
+ Wrapper around the basic Druid REST API operations using the
+ requests Python package. Handles the grunt work of building up
+ URLs, working with JSON, etc.
+ '''
+
+ def __init__(self, endpoint):
+ self.endpoint = endpoint
+ self.trace = False
+ self.session = requests.Session()
+
+ def enable_trace(self, flag=True):
+ self.trace = flag
+
+ def build_url(self, req, args=None) -> str:
+ """
+ Returns the full URL for a REST call given the relative request API and
+ optional parameters to fill placeholders within the request URL.
+
+ Parameters
+ ----------
+ req : str
+ relative URL, with optional {} placeholders
+
+ args : list
+ optional list of values to match {} placeholders
+ in the URL.
+ """
+ url = self.endpoint + req
+ if args is not None:
+ quoted = [quote(arg) for arg in args]
+ url = url.format(*quoted)
+ return url
+
+ def get(self, req, args=None, params=None, require_ok=True) -> requests.Request:
+ '''
+ Generic GET request to this service.
Review Comment:
```suggestion
Generic GET request to the service specified in `req`.
```
Not sure what "this" service is
##########
examples/quickstart/jupyter-notebooks/druidapi/rest.py:
##########
@@ -0,0 +1,180 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests
+from .util import dict_get, is_blank
+from urllib.parse import quote
+from .error import ClientError
+
+def check_error(response):
+ """
+ Raises a requests HttpError if the response code is not OK or Accepted.
+
+ If the response included a JSON payload, then the message is extracted
+ from that payload, else the message is from requests. The JSON
+ payload, if any, is returned in the json field of the error.
+ """
+ code = response.status_code
+ if code == requests.codes.ok or code == requests.codes.accepted:
+ return
+ error = None
+ json = None
+ try:
+ json = response.json()
+ except Exception:
+ # If we can't get the JSON, just move on, we'll figure
+ # things out another way.
+ pass
+ msg = dict_get(json, 'errorMessage')
+ if msg is None:
+ msg = dict_get(json, 'error')
+ if not is_blank(msg):
+ raise ClientError(msg)
+ if code == requests.codes.not_found and error is None:
+ error = "Not found"
+ if error is not None:
+ response.reason = error
+ try:
+ response.raise_for_status()
+ except Exception as e:
+ e.json = json
+ raise e
+
+class DruidRestClient:
+ '''
+ Wrapper around the basic Druid REST API operations using the
+ requests Python package. Handles the grunt work of building up
+ URLs, working with JSON, etc.
+ '''
+
+ def __init__(self, endpoint):
+ self.endpoint = endpoint
+ self.trace = False
+ self.session = requests.Session()
+
+ def enable_trace(self, flag=True):
+ self.trace = flag
+
+ def build_url(self, req, args=None) -> str:
+ """
+ Returns the full URL for a REST call given the relative request API and
+ optional parameters to fill placeholders within the request URL.
+
+ Parameters
+ ----------
+ req : str
+ relative URL, with optional {} placeholders
+
+ args : list
+ optional list of values to match {} placeholders
+ in the URL.
+ """
+ url = self.endpoint + req
+ if args is not None:
+ quoted = [quote(arg) for arg in args]
+ url = url.format(*quoted)
+ return url
+
+ def get(self, req, args=None, params=None, require_ok=True) -> requests.Request:
+ '''
+ Generic GET request to this service.
+
+ Parameters
+ ----------
+ req: str
+ The request URL without host, port or query string.
+ Example: `/status`
+
+ args: [str], default = None
+ Optional parameters to fill in to the URL.
+ Example: `/customer/{}`
+
+ params: dict, default = None
+ Optional map of query variables to send in
+ the URL. Query parameters are the name/values pairs
+ that appear after the `?` marker.
+
+ require_ok: bool, default = True
Review Comment:
```suggestion
req : str
The request URL without host, port or query string.
Example: `/status`
args : [str], default = None
Optional parameters to fill in to the URL.
Example: `/customer/{}`
params : dict, default = None
Optional map of query variables to send in
the URL. Query parameters are the name/values pairs
that appear after the `?` marker.
require_ok : bool, default = True
```
nit on spacing consistency
##########
examples/quickstart/jupyter-notebooks/druidapi/tasks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .consts import OVERLORD_BASE
+
+# Tasks
+REQ_TASKS = OVERLORD_BASE + '/tasks'
+REQ_POST_TASK = OVERLORD_BASE + '/task'
+REQ_GET_TASK = REQ_POST_TASK + '/{}'
+REQ_TASK_STATUS = REQ_GET_TASK + '/status'
+REQ_TASK_REPORTS = REQ_GET_TASK + '/reports'
+REQ_END_TASK = REQ_GET_TASK
+REQ_END_DS_TASKS = REQ_END_TASK + '/shutdownAllTasks'
+
+class TaskClient:
+ """
+ Client for task-related APIs. The APIs connect through the Router to
+ the Overlord.
+ """
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def tasks(self, state=None, table=None, type=None, max=None, created_time_interval=None):
+ '''
+ Retrieve list of tasks.
+
+ Parameters
+ ----------
+ state : str, default = None
+ Filter list of tasks by task state. Valid options are "running",
+ "complete", "waiting", and "pending". Constants are defined for
+ each of these in the `consts` file.
+ table : str, default = None
+ Return tasks filtered by Druid table (datasource).
+ created_time_interval : str, Default = None
+ Return tasks created within the specified interval.
+ max : int, default = None
+ Maximum number of "complete" tasks to return. Only applies when state is set to "complete".
+ type : str, default = None
+ filter tasks by task type.
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/tasks`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ params = {}
+ if state is not None:
+ params['state'] = state
+ if table is not None:
+ params['datasource'] = table
+ if type is not None:
+ params['type'] = type
+ if max is not None:
+ params['max'] = max
+ if created_time_interval is not None:
+ params['createdTimeInterval'] = created_time_interval
+ return self.client.get_json(REQ_TASKS, params=params)
+
+ def task(self, task_id):
+ """
+ Retrieve the "payload" of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
Review Comment:
```suggestion
The ID of the task to retrieve.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/tasks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .consts import OVERLORD_BASE
+
+# Tasks
+REQ_TASKS = OVERLORD_BASE + '/tasks'
+REQ_POST_TASK = OVERLORD_BASE + '/task'
+REQ_GET_TASK = REQ_POST_TASK + '/{}'
+REQ_TASK_STATUS = REQ_GET_TASK + '/status'
+REQ_TASK_REPORTS = REQ_GET_TASK + '/reports'
+REQ_END_TASK = REQ_GET_TASK
+REQ_END_DS_TASKS = REQ_END_TASK + '/shutdownAllTasks'
+
+class TaskClient:
+ """
+ Client for task-related APIs. The APIs connect through the Router to
+ the Overlord.
+ """
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def tasks(self, state=None, table=None, type=None, max=None, created_time_interval=None):
+ '''
+ Retrieve list of tasks.
+
+ Parameters
+ ----------
+ state : str, default = None
+ Filter list of tasks by task state. Valid options are "running",
+ "complete", "waiting", and "pending". Constants are defined for
+ each of these in the `consts` file.
+ table : str, default = None
+ Return tasks filtered by Druid table (datasource).
+ created_time_interval : str, Default = None
+ Return tasks created within the specified interval.
+ max : int, default = None
+ Maximum number of "complete" tasks to return. Only applies when state is set to "complete".
+ type : str, default = None
+ filter tasks by task type.
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/tasks`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ params = {}
+ if state is not None:
+ params['state'] = state
+ if table is not None:
+ params['datasource'] = table
+ if type is not None:
+ params['type'] = type
+ if max is not None:
+ params['max'] = max
+ if created_time_interval is not None:
+ params['createdTimeInterval'] = created_time_interval
+ return self.client.get_json(REQ_TASKS, params=params)
+
+ def task(self, task_id):
+ """
+ Retrieve the "payload" of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ """
+ return self.client.get_json(REQ_GET_TASK, args=[task_id])
+
+ def task_status(self, task_id):
+ '''
+ Retrieve the status of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
Review Comment:
```suggestion
Retrieves the status of a task.
Parameters
----------
task_id : str
The ID of the task to retrieve.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/tasks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .consts import OVERLORD_BASE
+
+# Tasks
+REQ_TASKS = OVERLORD_BASE + '/tasks'
+REQ_POST_TASK = OVERLORD_BASE + '/task'
+REQ_GET_TASK = REQ_POST_TASK + '/{}'
+REQ_TASK_STATUS = REQ_GET_TASK + '/status'
+REQ_TASK_REPORTS = REQ_GET_TASK + '/reports'
+REQ_END_TASK = REQ_GET_TASK
+REQ_END_DS_TASKS = REQ_END_TASK + '/shutdownAllTasks'
+
+class TaskClient:
+ """
+ Client for task-related APIs. The APIs connect through the Router to
+ the Overlord.
+ """
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def tasks(self, state=None, table=None, type=None, max=None, created_time_interval=None):
+ '''
+ Retrieve list of tasks.
+
+ Parameters
+ ----------
+ state : str, default = None
+ Filter list of tasks by task state. Valid options are "running",
+ "complete", "waiting", and "pending". Constants are defined for
+ each of these in the `consts` file.
+ table : str, default = None
+ Return tasks filtered by Druid table (datasource).
+ created_time_interval : str, Default = None
+ Return tasks created within the specified interval.
+ max : int, default = None
+ Maximum number of "complete" tasks to return. Only applies when state is set to "complete".
+ type : str, default = None
+ filter tasks by task type.
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/tasks`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ params = {}
+ if state is not None:
+ params['state'] = state
+ if table is not None:
+ params['datasource'] = table
+ if type is not None:
+ params['type'] = type
+ if max is not None:
+ params['max'] = max
+ if created_time_interval is not None:
+ params['createdTimeInterval'] = created_time_interval
+ return self.client.get_json(REQ_TASKS, params=params)
+
+ def task(self, task_id):
+ """
+ Retrieve the "payload" of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ """
+ return self.client.get_json(REQ_GET_TASK, args=[task_id])
+
+ def task_status(self, task_id):
+ '''
+ Retrieve the status of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}/status`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ return self.client.get_json(REQ_TASK_STATUS, args=[task_id])
+
+ def task_reports(self, task_id):
+ '''
+ Retrieve a task completion report for a task.
+ Only works for completed tasks.
Review Comment:
Could be helpful to say what happens when called on incomplete tasks. "Incomplete tasks raise an exception"? Or return null?
##########
examples/quickstart/jupyter-notebooks/druidapi/tasks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .consts import OVERLORD_BASE
+
+# Tasks
+REQ_TASKS = OVERLORD_BASE + '/tasks'
+REQ_POST_TASK = OVERLORD_BASE + '/task'
+REQ_GET_TASK = REQ_POST_TASK + '/{}'
+REQ_TASK_STATUS = REQ_GET_TASK + '/status'
+REQ_TASK_REPORTS = REQ_GET_TASK + '/reports'
+REQ_END_TASK = REQ_GET_TASK
+REQ_END_DS_TASKS = REQ_END_TASK + '/shutdownAllTasks'
+
+class TaskClient:
+ """
+ Client for task-related APIs. The APIs connect through the Router to
+ the Overlord.
+ """
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def tasks(self, state=None, table=None, type=None, max=None, created_time_interval=None):
+ '''
+ Retrieve list of tasks.
+
+ Parameters
+ ----------
+ state : str, default = None
+ Filter list of tasks by task state. Valid options are "running",
+ "complete", "waiting", and "pending". Constants are defined for
+ each of these in the `consts` file.
+ table : str, default = None
+ Return tasks filtered by Druid table (datasource).
+ created_time_interval : str, Default = None
+ Return tasks created within the specified interval.
+ max : int, default = None
+ Maximum number of "complete" tasks to return. Only applies when state is set to "complete".
+ type : str, default = None
+ filter tasks by task type.
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/tasks`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ params = {}
+ if state is not None:
+ params['state'] = state
+ if table is not None:
+ params['datasource'] = table
+ if type is not None:
+ params['type'] = type
+ if max is not None:
+ params['max'] = max
+ if created_time_interval is not None:
+ params['createdTimeInterval'] = created_time_interval
+ return self.client.get_json(REQ_TASKS, params=params)
+
+ def task(self, task_id):
+ """
+ Retrieve the "payload" of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ """
+ return self.client.get_json(REQ_GET_TASK, args=[task_id])
+
+ def task_status(self, task_id):
+ '''
+ Retrieve the status of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}/status`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ return self.client.get_json(REQ_TASK_STATUS, args=[task_id])
+
+ def task_reports(self, task_id):
+ '''
+ Retrieve a task completion report for a task.
Review Comment:
```suggestion
Retrieves a task completion report for a task.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/util.py:
##########
@@ -0,0 +1,30 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+def is_blank(s):
+ """
+ Returns True if the given string is None or blank (after stripping spaces),
+ False otherwise.
Review Comment:
```suggestion
Strips spaces from a string and checks for an empty string.
Returns `True` if the given string is None or blank, `False` otherwise.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/tasks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .consts import OVERLORD_BASE
+
+# Tasks
+REQ_TASKS = OVERLORD_BASE + '/tasks'
+REQ_POST_TASK = OVERLORD_BASE + '/task'
+REQ_GET_TASK = REQ_POST_TASK + '/{}'
+REQ_TASK_STATUS = REQ_GET_TASK + '/status'
+REQ_TASK_REPORTS = REQ_GET_TASK + '/reports'
+REQ_END_TASK = REQ_GET_TASK
+REQ_END_DS_TASKS = REQ_END_TASK + '/shutdownAllTasks'
+
+class TaskClient:
+ """
+ Client for task-related APIs. The APIs connect through the Router to
+ the Overlord.
+ """
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def tasks(self, state=None, table=None, type=None, max=None, created_time_interval=None):
+ '''
+ Retrieve list of tasks.
+
+ Parameters
+ ----------
+ state : str, default = None
+ Filter list of tasks by task state. Valid options are "running",
+ "complete", "waiting", and "pending". Constants are defined for
+ each of these in the `consts` file.
+ table : str, default = None
+ Return tasks filtered by Druid table (datasource).
+ created_time_interval : str, Default = None
+ Return tasks created within the specified interval.
+ max : int, default = None
+ Maximum number of "complete" tasks to return. Only applies when state is set to "complete".
+ type : str, default = None
+ filter tasks by task type.
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/tasks`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ params = {}
+ if state is not None:
+ params['state'] = state
+ if table is not None:
+ params['datasource'] = table
+ if type is not None:
+ params['type'] = type
+ if max is not None:
+ params['max'] = max
+ if created_time_interval is not None:
+ params['createdTimeInterval'] = created_time_interval
+ return self.client.get_json(REQ_TASKS, params=params)
+
+ def task(self, task_id):
+ """
+ Retrieve the "payload" of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ """
+ return self.client.get_json(REQ_GET_TASK, args=[task_id])
+
+ def task_status(self, task_id):
+ '''
+ Retrieve the status of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}/status`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ return self.client.get_json(REQ_TASK_STATUS, args=[task_id])
+
+ def task_reports(self, task_id):
+ '''
+ Retrieve a task completion report for a task.
+ Only works for completed tasks.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}/reports`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ return self.client.get_json(REQ_TASK_REPORTS, args=[task_id])
+
+ def submit_task(self, payload):
+ """
+ Submit a task or supervisor specs to the Overlord.
+
+ Returns the taskId of the submitted task.
+
+ Parameters
+ ----------
+ payload : object
+ The task object. Serialized to JSON.
+
+ Reference
+ ---------
+ `POST /druid/indexer/v1/task`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#post-5
+ """
+ return self.client.post_json(REQ_POST_TASK, payload)
+
+ def shut_down_task(self, task_id):
+ """
+ Shuts down a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to shut down
Review Comment:
```suggestion
The ID of the task to shut down.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/tasks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .consts import OVERLORD_BASE
+
+# Tasks
+REQ_TASKS = OVERLORD_BASE + '/tasks'
+REQ_POST_TASK = OVERLORD_BASE + '/task'
+REQ_GET_TASK = REQ_POST_TASK + '/{}'
+REQ_TASK_STATUS = REQ_GET_TASK + '/status'
+REQ_TASK_REPORTS = REQ_GET_TASK + '/reports'
+REQ_END_TASK = REQ_GET_TASK
+REQ_END_DS_TASKS = REQ_END_TASK + '/shutdownAllTasks'
+
+class TaskClient:
+ """
+ Client for task-related APIs. The APIs connect through the Router to
+ the Overlord.
+ """
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def tasks(self, state=None, table=None, type=None, max=None, created_time_interval=None):
+ '''
+ Retrieve list of tasks.
+
+ Parameters
+ ----------
+ state : str, default = None
+ Filter list of tasks by task state. Valid options are "running",
+ "complete", "waiting", and "pending". Constants are defined for
+ each of these in the `consts` file.
+ table : str, default = None
+ Return tasks filtered by Druid table (datasource).
+ created_time_interval : str, Default = None
+ Return tasks created within the specified interval.
+ max : int, default = None
+ Maximum number of "complete" tasks to return. Only applies when state is set to "complete".
+ type : str, default = None
+ filter tasks by task type.
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/tasks`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ params = {}
+ if state is not None:
+ params['state'] = state
+ if table is not None:
+ params['datasource'] = table
+ if type is not None:
+ params['type'] = type
+ if max is not None:
+ params['max'] = max
+ if created_time_interval is not None:
+ params['createdTimeInterval'] = created_time_interval
+ return self.client.get_json(REQ_TASKS, params=params)
+
+ def task(self, task_id):
+ """
+ Retrieve the "payload" of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ """
+ return self.client.get_json(REQ_GET_TASK, args=[task_id])
+
+ def task_status(self, task_id):
+ '''
+ Retrieve the status of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}/status`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ return self.client.get_json(REQ_TASK_STATUS, args=[task_id])
+
+ def task_reports(self, task_id):
+ '''
+ Retrieve a task completion report for a task.
+ Only works for completed tasks.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}/reports`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ return self.client.get_json(REQ_TASK_REPORTS, args=[task_id])
+
+ def submit_task(self, payload):
+ """
+ Submit a task or supervisor specs to the Overlord.
Review Comment:
```suggestion
Submits a task or supervisor spec to the Overlord.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/tasks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .consts import OVERLORD_BASE
+
+# Tasks
+REQ_TASKS = OVERLORD_BASE + '/tasks'
+REQ_POST_TASK = OVERLORD_BASE + '/task'
+REQ_GET_TASK = REQ_POST_TASK + '/{}'
+REQ_TASK_STATUS = REQ_GET_TASK + '/status'
+REQ_TASK_REPORTS = REQ_GET_TASK + '/reports'
+REQ_END_TASK = REQ_GET_TASK
+REQ_END_DS_TASKS = REQ_END_TASK + '/shutdownAllTasks'
+
+class TaskClient:
+ """
+ Client for task-related APIs. The APIs connect through the Router to
+ the Overlord.
+ """
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def tasks(self, state=None, table=None, type=None, max=None, created_time_interval=None):
+ '''
+ Retrieve list of tasks.
+
+ Parameters
+ ----------
+ state : str, default = None
+ Filter list of tasks by task state. Valid options are "running",
+ "complete", "waiting", and "pending". Constants are defined for
+ each of these in the `consts` file.
+ table : str, default = None
+ Return tasks filtered by Druid table (datasource).
+ created_time_interval : str, Default = None
+ Return tasks created within the specified interval.
+ max : int, default = None
+ Maximum number of "complete" tasks to return. Only applies when state is set to "complete".
+ type : str, default = None
+ filter tasks by task type.
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/tasks`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ params = {}
+ if state is not None:
+ params['state'] = state
+ if table is not None:
+ params['datasource'] = table
+ if type is not None:
+ params['type'] = type
+ if max is not None:
+ params['max'] = max
+ if created_time_interval is not None:
+ params['createdTimeInterval'] = created_time_interval
+ return self.client.get_json(REQ_TASKS, params=params)
+
+ def task(self, task_id):
+ """
+ Retrieve the "payload" of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ """
+ return self.client.get_json(REQ_GET_TASK, args=[task_id])
+
+ def task_status(self, task_id):
+ '''
+ Retrieve the status of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}/status`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ return self.client.get_json(REQ_TASK_STATUS, args=[task_id])
+
+ def task_reports(self, task_id):
+ '''
+ Retrieve a task completion report for a task.
+ Only works for completed tasks.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}/reports`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ return self.client.get_json(REQ_TASK_REPORTS, args=[task_id])
+
+ def submit_task(self, payload):
+ """
+ Submit a task or supervisor specs to the Overlord.
+
+ Returns the taskId of the submitted task.
+
+ Parameters
+ ----------
+ payload : object
+ The task object. Serialized to JSON.
+
+ Reference
+ ---------
+ `POST /druid/indexer/v1/task`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#post-5
+ """
+ return self.client.post_json(REQ_POST_TASK, payload)
+
+ def shut_down_task(self, task_id):
+ """
+ Shuts down a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to shut down
+
+ Reference
+ ---------
+ `POST /druid/indexer/v1/task/{taskId}/shutdown`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#post-5
+ """
+ return self.client.post_json(REQ_END_TASK, args=[task_id])
+
+ def shut_down_tasks_for(self, table):
+ """
+ Shuts down all tasks for a table (data source).
+
+ Parameters
+ ----------
+ table : str
+ The name of the table (data source).
Review Comment:
```suggestion
Shuts down all tasks for a table (datasource).
Parameters
----------
table : str
The name of the table (datasource).
```
##########
examples/quickstart/jupyter-notebooks/druidapi/tasks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .consts import OVERLORD_BASE
+
+# Tasks
+REQ_TASKS = OVERLORD_BASE + '/tasks'
+REQ_POST_TASK = OVERLORD_BASE + '/task'
+REQ_GET_TASK = REQ_POST_TASK + '/{}'
+REQ_TASK_STATUS = REQ_GET_TASK + '/status'
+REQ_TASK_REPORTS = REQ_GET_TASK + '/reports'
+REQ_END_TASK = REQ_GET_TASK
+REQ_END_DS_TASKS = REQ_END_TASK + '/shutdownAllTasks'
+
+class TaskClient:
+ """
+ Client for task-related APIs. The APIs connect through the Router to
+ the Overlord.
+ """
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def tasks(self, state=None, table=None, type=None, max=None, created_time_interval=None):
+ '''
+ Retrieve list of tasks.
+
+ Parameters
+ ----------
+ state : str, default = None
+ Filter list of tasks by task state. Valid options are "running",
+ "complete", "waiting", and "pending". Constants are defined for
+ each of these in the `consts` file.
+ table : str, default = None
+ Return tasks filtered by Druid table (datasource).
+ created_time_interval : str, Default = None
+ Return tasks created within the specified interval.
+ max : int, default = None
+ Maximum number of "complete" tasks to return. Only applies when state is set to "complete".
+ type : str, default = None
+ filter tasks by task type.
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/tasks`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ params = {}
+ if state is not None:
+ params['state'] = state
+ if table is not None:
+ params['datasource'] = table
+ if type is not None:
+ params['type'] = type
+ if max is not None:
+ params['max'] = max
+ if created_time_interval is not None:
+ params['createdTimeInterval'] = created_time_interval
+ return self.client.get_json(REQ_TASKS, params=params)
+
+ def task(self, task_id):
+ """
+ Retrieve the "payload" of a task.
Review Comment:
```suggestion
Retrieves the "payload" of a task.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/sql.py:
##########
@@ -0,0 +1,690 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time, requests
+from . import consts, display
+from .consts import ROUTER_BASE
+from .util import is_blank, dict_get
+from .error import DruidError, ClientError
+
+REQ_ROUTER_QUERY = ROUTER_BASE
+REQ_ROUTER_SQL = ROUTER_BASE + '/sql'
+REQ_ROUTER_SQL_TASK = REQ_ROUTER_SQL + '/task'
+
+class SqlRequest:
+
+ def __init__(self, query_client, sql):
+ self.query_client = query_client
+ self.sql = sql
+ self.context = None
+ self.params = None
+ self.header = False
+ self.format = consts.SQL_OBJECT
+ self.headers = None
+ self.types = None
+ self.sqlTypes = None
+
+ def with_format(self, result_format):
+ self.format = result_format
+ return self
+
+ def with_headers(self, sqlTypes=False, druidTypes=False):
+ self.headers = True
+ self.types = druidTypes
+ self.sqlTypes = sqlTypes
+ return self
+
+ def with_context(self, context):
+ if self.context is None:
+ self.context = context
+ else:
+ self.context.update(context)
+ return self
+
+ def with_parameters(self, params):
+ '''
+ Set the array of parameters. Parameters must each be a map of 'type'/'value' pairs:
+ {'type': the_type, 'value': the_value}. The type must be a valid SQL type
+ (in upper case). See the consts module for a list.
+ '''
+ if self.params is None:
+ self.params = params
+ else:
+ self.params.update(params)
+ return self
+
+ def add_parameter(self, value):
+ '''
+ Add one parameter value. Infers the type of the parameter from the Python type.
+ '''
+ if value is None:
+ raise ClientError("Druid does not support null parameter values")
+ data_type = None
+ value_type = type(value)
+ if value_type is str:
+ data_type = consts.SQL_VARCHAR_TYPE
+ elif value_type is int:
+ data_type = consts.SQL_BIGINT_TYPE
+ elif value_type is float:
+ data_type = consts.SQL_DOUBLE_TYPE
+ elif value_type is list:
+ data_type = consts.SQL_ARRAY_TYPE
+ else:
+ raise ClientError("Unsupported value type")
+ if self.params is None:
+ self.params = []
+ self.params.append({'type': data_type, 'value': value})
+
+ def response_header(self):
+ self.header = True
+ return self
+
+ def request_headers(self, headers):
+ self.headers = headers
+ return self
+
+ def to_request(self):
+ query_obj = {"query": self.sql}
+ if self.context is not None and len(self.context) > 0:
+ query_obj['context'] = self.context
+ if self.params is not None and len(self.params) > 0:
+ query_obj['parameters'] = self.params
+ if self.header:
+ query_obj['header'] = True
+ if self.result_format is not None:
+ query_obj['resultFormat'] = self.format
+ if self.sqlTypes:
+ query_obj['sqlTypesHeader'] = self.sqlTypes
+ if self.types:
+ query_obj['typesHeader'] = self.types
+ return query_obj
+
+ def result_format(self):
+ return self.format.lower()
+
+ def run(self):
+ return self.query_client.sql_query(self)
+
+def parse_rows(fmt, context, results):
+ if fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ rows = results['results']
+ elif fmt == consts.SQL_ARRAY:
+ rows = results
+ else:
+ return results
+ if not context.get('headers', False):
+ return rows
+ header_size = 1
+ if context.get('sqlTypesHeader', False):
+ header_size += 1
+ if context.get('typesHeader', False):
+ header_size += 1
+ return rows[header_size:]
+
+def label_non_null_cols(results):
+ if results is None or len(results) == 0:
+ return []
+ is_null = {}
+ for key in results[0].keys():
+ is_null[key] = True
+ for row in results:
+ for key, value in row.items():
+ if type(value) == str:
+ if value != '':
+ is_null[key] = False
+ elif type(value) == float:
+ if value != 0.0:
+ is_null[key] = False
+ elif value is not None:
+ is_null[key] = False
+ return is_null
+
+def filter_null_cols(results):
+ '''
+ Filter columns from a Druid result set by removing all null-like
+ columns. A column is considered null if all values for that column
+ are null. A value is null if it is either a JSON null, an empty
+ string, or a numeric 0. All rows are preserved, as is the order
+ of the remaining columns.
+ '''
+ if results is None or len(results) == 0:
+ return results
+ is_null = label_non_null_cols(results)
+ revised = []
+ for row in results:
+ new_row = {}
+ for key, value in row.items():
+ if is_null[key]:
+ continue
+ new_row[key] = value
+ revised.append(new_row)
+ return revised
+
+def parse_object_schema(results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ row = results[0]
+ for k, v in row.items():
+ druid_type = None
+ sql_type = None
+ if type(v) is str:
+ druid_type = consts.DRUID_STRING_TYPE
+ sql_type = consts.SQL_VARCHAR_TYPE
+ elif type(v) is int or type(v) is float:
+ druid_type = consts.DRUID_LONG_TYPE
+ sql_type = consts.SQL_BIGINT_TYPE
+ schema.append(ColumnSchema(k, sql_type, druid_type))
+ return schema
+
+def parse_array_schema(context, results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ has_headers = context.get(consts.HEADERS_KEY, False)
+ if not has_headers:
+ return schema
+ has_sql_types = context.get(consts.SQL_TYPES_HEADERS_KEY, False)
+ has_druid_types = context.get(consts.DRUID_TYPE_HEADERS_KEY, False)
+ size = len(results[0])
+ for i in range(size):
+ druid_type = None
+ if has_druid_types:
+ druid_type = results[1][i]
+ sql_type = None
+ if has_sql_types:
+ sql_type = results[2][i]
+ schema.append(ColumnSchema(results[0][i], sql_type, druid_type))
+ return schema
+
+def parse_schema(fmt, context, results):
+ if fmt == consts.SQL_OBJECT:
+ return parse_object_schema(results)
+ elif fmt == consts.SQL_ARRAY or fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ return parse_array_schema(context, results)
+ else:
+ return []
+
+def is_response_ok(http_response):
+ code = http_response.status_code
+ return code == requests.codes.ok or code == requests.codes.accepted
+
+class ColumnSchema:
+
+ def __init__(self, name, sql_type, druid_type):
+ self.name = name
+ self.sql_type = sql_type
+ self.druid_type = druid_type
+
+ def __str__(self):
+ return "{{name={}, SQL type={}, Druid type={}}}".format(self.name, self.sql_type, self.druid_type)
+
+class SqlQueryResult:
+ """
+ Defines the core protocol for Druid SQL queries.
+ """
+
+ def __init__(self, request, response):
+ self.http_response = response
+ self._json = None
+ self._rows = None
+ self._schema = None
+ self.request = request
+ self._error = None
+ self._id = None
+ if not is_response_ok(response):
+ try:
+ self._error = response.json()
+ except Exception:
+ self._error = response.text
+ if self._error is None or len(self._error) == 0:
+ self._error = "Failed with HTTP status {}".format(response.status_code)
+ try:
+ self._id = self.http_response.headers['X-Druid-SQL-Query-Id']
+ except KeyError:
+ self._error = "Query returned no query ID"
+
+ def result_format(self):
+ return self.request.result_format()
+
+ def ok(self):
+ """
+ Reports if the query succeeded.
+
+ The query rows and schema are available only if ok() returns True.
+ """
+ return is_response_ok(self.http_response)
+
+ def error_msg(self):
+ err = self.error()
+ if err is None:
+ return "unknown"
+ if type(err) is str:
+ return err
+ msg = err.get("error")
+ text = err.get("errorMessage")
+ if msg is None and text is None:
+ return "unknown"
+ if msg is None:
+ return text
+ if text is None:
+ return msg
+ return msg + ": " + text
+
+ def id(self):
+ """
+ Returns the unique identifier for the query.
+ """
+ return self._id
+
+ def non_null(self):
+ if not self.ok():
+ return None
+ if self.result_format() != consts.SQL_OBJECT:
+ return None
+ return filter_null_cols(self.rows())
+
+ def as_array(self):
+ if self.result_format() == consts.SQL_OBJECT:
+ rows = []
+ for obj in self.rows():
+ rows.append([v for v in obj.values()])
+ return rows
+ else:
+ return self.rows()
+
+ def error(self):
+ """
+ If the query fails, returns the error, if any provided by Druid.
+ """
+ if self.ok():
+ return None
+ if self._error is not None:
+ return self._error
+ if self.http_response is None:
+ return { "error": "unknown"}
+ if is_response_ok(self.http_response):
+ return None
+ return {"error": "HTTP {}".format(self.http_response.status_code)}
+
+ def json(self):
+ if not self.ok():
+ return None
+ if self._json is None:
+ self._json = self.http_response.json()
+ return self._json
+
+ def rows(self):
+ """
+ Returns the rows of data for the query.
+
+ Druid supports many data formats. The method makes its best
+ attempt to map the format into an array of rows of some sort.
+ """
+ if self._rows is None:
+ json = self.json()
+ if json is None:
+ return self.http_response.text
+ self._rows = parse_rows(self.result_format(), self.request.context, json)
+ return self._rows
+
+ def schema(self):
+ """
+ Returns the data schema as a list of ColumnSchema objects.
+
+ Druid supports many data formats, not all of them provide
+ schema information. This method makes its best attempt to
+ extract the schema from the query results.
+ """
+ if self._schema is None:
+ self._schema = parse_schema(self.result_format(), self.request.context, self.json())
+ return self._schema
+
+ def show(self, non_null=False):
+ data = None
+ if non_null:
+ data = self.non_null()
+ if data is None:
+ data = self.as_array()
+ if data is None or len(data) == 0:
+ display.display.show_message("Query returned no results")
+ return
+ disp = display.display.table()
+ disp.headers([c.name for c in self.schema()])
+ disp.show(data)
+
+ def show_schema(self):
+ disp = display.display.table()
+ disp.headers(['Name', 'SQL Type', 'Druid Type'])
+ data = []
+ for c in self.schema():
+ data.append([c.name, c.sql_type, c.druid_type])
+ disp.show(data)
+
+class QueryTaskResult:
+
+ def __init__(self, request, response):
+ self._request = request
+ self.http_response = response
+ self._status = None
+ self._results = None
+ self._details = None
+ self._schema = None
+ self._rows = None
+ self._reports = None
+ self._schema = None
+ self._results = None
+ self._error = None
+ self._id = None
+ if not is_response_ok(response):
+ self._state = consts.FAILED_STATE
+ try:
+ self._error = response.json()
+ except Exception:
+ self._error = response.text
+ if self._error is None or len(self._error) == 0:
+ self._error = "Failed with HTTP status {}".format(response.status_code)
+ return
+
+ # Typical response:
+ # {'taskId': '6f7b514a446d4edc9d26a24d4bd03ade_fd8e242b-7d93-431d-b65b-2a512116924c_bjdlojgj',
+ # 'state': 'RUNNING'}
+ self.response_obj = response.json()
+ self._id = self.response_obj['taskId']
+ self._state = self.response_obj['state']
+
+ def ok(self):
+ """
+ Reports if the query succeeded.
+
+ The query rows and schema are available only if ok() returns True.
+ """
+ return self._error is None
+
+ def id(self):
+ return self._id
+
+ def _tasks(self):
+ return self._request.query_client.druid_client.tasks()
+
+ def status(self):
+ """
+ Polls Druid for an update on the query run status.
+ """
+ self.check_valid()
+ # Example:
+ # {'task': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'status': {
+ # 'id': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'groupId': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'type': 'talaria0', 'createdTime': '2022-04-28T23:19:50.331Z',
+ # 'queueInsertionTime': '1970-01-01T00:00:00.000Z',
+ # 'statusCode': 'RUNNING', 'status': 'RUNNING', 'runnerStatusCode': 'PENDING',
+ # 'duration': -1, 'location': {'host': None, 'port': -1, 'tlsPort': -1},
+ # 'dataSource': 'w000', 'errorMsg': None}}
+ self._status = self._tasks().task_status(self._id)
+ self._state = self._status['status']['status']
+ if self._state == consts.FAILED_STATE:
+ self._error = self._status['status']['errorMsg']
+ return self._status
+
+ def done(self):
+ """
+ Reports if the query is done: succeeded or failed.
Review Comment:
```suggestion
Reports the state of a query, either `SUCCESS` or `FAILED`.
```
or
```suggestion
Reports the state of a query, whether it succeeded or failed.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/status.py:
##########
@@ -0,0 +1,99 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time
+
+STATUS_BASE = "/status"
+REQ_STATUS = STATUS_BASE
+REQ_HEALTH = STATUS_BASE + "/health"
+REQ_PROPERTIES = STATUS_BASE + "/properties"
+REQ_IN_CLUSTER = STATUS_BASE + "/selfDiscovered/status"
+
+ROUTER_BASE = '/druid/router/v1'
+REQ_BROKERS = ROUTER_BASE + '/brokers'
+
+class StatusClient:
+ '''
+ Client for status APIs. These APIs are available on all nodes.
+ If used with the router, they report the status of just the router.
+ '''
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ #-------- Common --------
+
+ def status(self):
+ """
+ Returns the Druid version, loaded extensions, memory used, total memory
+ and other useful information about the process.
Review Comment:
```suggestion
and other useful information about Druid.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/status.py:
##########
@@ -0,0 +1,99 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time
+
+STATUS_BASE = "/status"
+REQ_STATUS = STATUS_BASE
+REQ_HEALTH = STATUS_BASE + "/health"
+REQ_PROPERTIES = STATUS_BASE + "/properties"
+REQ_IN_CLUSTER = STATUS_BASE + "/selfDiscovered/status"
+
+ROUTER_BASE = '/druid/router/v1'
+REQ_BROKERS = ROUTER_BASE + '/brokers'
+
+class StatusClient:
+ '''
+ Client for status APIs. These APIs are available on all nodes.
+ If used with the router, they report the status of just the router.
+ '''
Review Comment:
```suggestion
"""
Client for status APIs. These APIs are available on all nodes.
If used with the Router, they report the status of just the Router.
"""
```
##########
examples/quickstart/jupyter-notebooks/druidapi/sql.py:
##########
@@ -0,0 +1,690 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time, requests
+from . import consts, display
+from .consts import ROUTER_BASE
+from .util import is_blank, dict_get
+from .error import DruidError, ClientError
+
+REQ_ROUTER_QUERY = ROUTER_BASE
+REQ_ROUTER_SQL = ROUTER_BASE + '/sql'
+REQ_ROUTER_SQL_TASK = REQ_ROUTER_SQL + '/task'
+
+class SqlRequest:
+
+ def __init__(self, query_client, sql):
+ self.query_client = query_client
+ self.sql = sql
+ self.context = None
+ self.params = None
+ self.header = False
+ self.format = consts.SQL_OBJECT
+ self.headers = None
+ self.types = None
+ self.sqlTypes = None
+
+ def with_format(self, result_format):
+ self.format = result_format
+ return self
+
+ def with_headers(self, sqlTypes=False, druidTypes=False):
+ self.headers = True
+ self.types = druidTypes
+ self.sqlTypes = sqlTypes
+ return self
+
+ def with_context(self, context):
+ if self.context is None:
+ self.context = context
+ else:
+ self.context.update(context)
+ return self
+
+ def with_parameters(self, params):
+ '''
+ Set the array of parameters. Parameters must each be a map of 'type'/'value' pairs:
+ {'type': the_type, 'value': the_value}. The type must be a valid SQL type
+ (in upper case). See the consts module for a list.
+ '''
+ if self.params is None:
+ self.params = params
+ else:
+ self.params.update(params)
+ return self
+
+ def add_parameter(self, value):
+ '''
+ Add one parameter value. Infers the type of the parameter from the Python type.
+ '''
+ if value is None:
+ raise ClientError("Druid does not support null parameter values")
+ data_type = None
+ value_type = type(value)
+ if value_type is str:
+ data_type = consts.SQL_VARCHAR_TYPE
+ elif value_type is int:
+ data_type = consts.SQL_BIGINT_TYPE
+ elif value_type is float:
+ data_type = consts.SQL_DOUBLE_TYPE
+ elif value_type is list:
+ data_type = consts.SQL_ARRAY_TYPE
+ else:
+ raise ClientError("Unsupported value type")
+ if self.params is None:
+ self.params = []
+ self.params.append({'type': data_type, 'value': value})
+
+ def response_header(self):
+ self.header = True
+ return self
+
+ def request_headers(self, headers):
+ self.headers = headers
+ return self
+
+ def to_request(self):
+ query_obj = {"query": self.sql}
+ if self.context is not None and len(self.context) > 0:
+ query_obj['context'] = self.context
+ if self.params is not None and len(self.params) > 0:
+ query_obj['parameters'] = self.params
+ if self.header:
+ query_obj['header'] = True
+ if self.result_format is not None:
+ query_obj['resultFormat'] = self.format
+ if self.sqlTypes:
+ query_obj['sqlTypesHeader'] = self.sqlTypes
+ if self.types:
+ query_obj['typesHeader'] = self.types
+ return query_obj
+
+ def result_format(self):
+ return self.format.lower()
+
+ def run(self):
+ return self.query_client.sql_query(self)
+
+def parse_rows(fmt, context, results):
+ if fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ rows = results['results']
+ elif fmt == consts.SQL_ARRAY:
+ rows = results
+ else:
+ return results
+ if not context.get('headers', False):
+ return rows
+ header_size = 1
+ if context.get('sqlTypesHeader', False):
+ header_size += 1
+ if context.get('typesHeader', False):
+ header_size += 1
+ return rows[header_size:]
+
+def label_non_null_cols(results):
+ if results is None or len(results) == 0:
+ return []
+ is_null = {}
+ for key in results[0].keys():
+ is_null[key] = True
+ for row in results:
+ for key, value in row.items():
+ if type(value) == str:
+ if value != '':
+ is_null[key] = False
+ elif type(value) == float:
+ if value != 0.0:
+ is_null[key] = False
+ elif value is not None:
+ is_null[key] = False
+ return is_null
+
+def filter_null_cols(results):
+ '''
+ Filter columns from a Druid result set by removing all null-like
+ columns. A column is considered null if all values for that column
+ are null. A value is null if it is either a JSON null, an empty
+ string, or a numeric 0. All rows are preserved, as is the order
+ of the remaining columns.
+ '''
+ if results is None or len(results) == 0:
+ return results
+ is_null = label_non_null_cols(results)
+ revised = []
+ for row in results:
+ new_row = {}
+ for key, value in row.items():
+ if is_null[key]:
+ continue
+ new_row[key] = value
+ revised.append(new_row)
+ return revised
+
+def parse_object_schema(results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ row = results[0]
+ for k, v in row.items():
+ druid_type = None
+ sql_type = None
+ if type(v) is str:
+ druid_type = consts.DRUID_STRING_TYPE
+ sql_type = consts.SQL_VARCHAR_TYPE
+ elif type(v) is int or type(v) is float:
+ druid_type = consts.DRUID_LONG_TYPE
+ sql_type = consts.SQL_BIGINT_TYPE
+ schema.append(ColumnSchema(k, sql_type, druid_type))
+ return schema
+
+def parse_array_schema(context, results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ has_headers = context.get(consts.HEADERS_KEY, False)
+ if not has_headers:
+ return schema
+ has_sql_types = context.get(consts.SQL_TYPES_HEADERS_KEY, False)
+ has_druid_types = context.get(consts.DRUID_TYPE_HEADERS_KEY, False)
+ size = len(results[0])
+ for i in range(size):
+ druid_type = None
+ if has_druid_types:
+ druid_type = results[1][i]
+ sql_type = None
+ if has_sql_types:
+ sql_type = results[2][i]
+ schema.append(ColumnSchema(results[0][i], sql_type, druid_type))
+ return schema
+
+def parse_schema(fmt, context, results):
+ if fmt == consts.SQL_OBJECT:
+ return parse_object_schema(results)
+ elif fmt == consts.SQL_ARRAY or fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ return parse_array_schema(context, results)
+ else:
+ return []
+
+def is_response_ok(http_response):
+ code = http_response.status_code
+ return code == requests.codes.ok or code == requests.codes.accepted
+
+class ColumnSchema:
+
+ def __init__(self, name, sql_type, druid_type):
+ self.name = name
+ self.sql_type = sql_type
+ self.druid_type = druid_type
+
+ def __str__(self):
+ return "{{name={}, SQL type={}, Druid type={}}}".format(self.name, self.sql_type, self.druid_type)
+
+class SqlQueryResult:
+ """
+ Defines the core protocol for Druid SQL queries.
+ """
+
+ def __init__(self, request, response):
+ self.http_response = response
+ self._json = None
+ self._rows = None
+ self._schema = None
+ self.request = request
+ self._error = None
+ self._id = None
+ if not is_response_ok(response):
+ try:
+ self._error = response.json()
+ except Exception:
+ self._error = response.text
+ if self._error is None or len(self._error) == 0:
+ self._error = "Failed with HTTP status {}".format(response.status_code)
+ try:
+ self._id = self.http_response.headers['X-Druid-SQL-Query-Id']
+ except KeyError:
+ self._error = "Query returned no query ID"
+
+ def result_format(self):
+ return self.request.result_format()
+
+ def ok(self):
+ """
+ Reports if the query succeeded.
+
+ The query rows and schema are available only if ok() returns True.
+ """
+ return is_response_ok(self.http_response)
+
+ def error_msg(self):
+ err = self.error()
+ if err is None:
+ return "unknown"
+ if type(err) is str:
+ return err
+ msg = err.get("error")
+ text = err.get("errorMessage")
+ if msg is None and text is None:
+ return "unknown"
+ if msg is None:
+ return text
+ if text is None:
+ return msg
+ return msg + ": " + text
+
+ def id(self):
+ """
+ Returns the unique identifier for the query.
+ """
+ return self._id
+
+ def non_null(self):
+ if not self.ok():
+ return None
+ if self.result_format() != consts.SQL_OBJECT:
+ return None
+ return filter_null_cols(self.rows())
+
+ def as_array(self):
+ if self.result_format() == consts.SQL_OBJECT:
+ rows = []
+ for obj in self.rows():
+ rows.append([v for v in obj.values()])
+ return rows
+ else:
+ return self.rows()
+
+ def error(self):
+ """
+ If the query fails, returns the error, if any provided by Druid.
+ """
+ if self.ok():
+ return None
+ if self._error is not None:
+ return self._error
+ if self.http_response is None:
+ return { "error": "unknown"}
+ if is_response_ok(self.http_response):
+ return None
+ return {"error": "HTTP {}".format(self.http_response.status_code)}
+
+ def json(self):
+ if not self.ok():
+ return None
+ if self._json is None:
+ self._json = self.http_response.json()
+ return self._json
+
+ def rows(self):
+ """
+ Returns the rows of data for the query.
+
+ Druid supports many data formats. The method makes its best
+ attempt to map the format into an array of rows of some sort.
+ """
+ if self._rows is None:
+ json = self.json()
+ if json is None:
+ return self.http_response.text
+ self._rows = parse_rows(self.result_format(), self.request.context, json)
+ return self._rows
+
+ def schema(self):
+ """
+ Returns the data schema as a list of ColumnSchema objects.
+
+ Druid supports many data formats, not all of them provide
+ schema information. This method makes its best attempt to
+ extract the schema from the query results.
+ """
+ if self._schema is None:
+ self._schema = parse_schema(self.result_format(), self.request.context, self.json())
+ return self._schema
+
+ def show(self, non_null=False):
+ data = None
+ if non_null:
+ data = self.non_null()
+ if data is None:
+ data = self.as_array()
+ if data is None or len(data) == 0:
+ display.display.show_message("Query returned no results")
+ return
+ disp = display.display.table()
+ disp.headers([c.name for c in self.schema()])
+ disp.show(data)
+
+ def show_schema(self):
+ disp = display.display.table()
+ disp.headers(['Name', 'SQL Type', 'Druid Type'])
+ data = []
+ for c in self.schema():
+ data.append([c.name, c.sql_type, c.druid_type])
+ disp.show(data)
+
+class QueryTaskResult:
+
+ def __init__(self, request, response):
+ self._request = request
+ self.http_response = response
+ self._status = None
+ self._results = None
+ self._details = None
+ self._schema = None
+ self._rows = None
+ self._reports = None
+ self._schema = None
+ self._results = None
+ self._error = None
+ self._id = None
+ if not is_response_ok(response):
+ self._state = consts.FAILED_STATE
+ try:
+ self._error = response.json()
+ except Exception:
+ self._error = response.text
+ if self._error is None or len(self._error) == 0:
+ self._error = "Failed with HTTP status {}".format(response.status_code)
+ return
+
+ # Typical response:
+ # {'taskId': '6f7b514a446d4edc9d26a24d4bd03ade_fd8e242b-7d93-431d-b65b-2a512116924c_bjdlojgj',
+ # 'state': 'RUNNING'}
+ self.response_obj = response.json()
+ self._id = self.response_obj['taskId']
+ self._state = self.response_obj['state']
+
+ def ok(self):
+ """
+ Reports if the query succeeded.
+
+ The query rows and schema are available only if ok() returns True.
+ """
+ return self._error is None
+
+ def id(self):
+ return self._id
+
+ def _tasks(self):
+ return self._request.query_client.druid_client.tasks()
+
+ def status(self):
+ """
+ Polls Druid for an update on the query run status.
+ """
+ self.check_valid()
+ # Example:
+ # {'task': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'status': {
+ # 'id': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'groupId': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'type': 'talaria0', 'createdTime': '2022-04-28T23:19:50.331Z',
+ # 'queueInsertionTime': '1970-01-01T00:00:00.000Z',
+ # 'statusCode': 'RUNNING', 'status': 'RUNNING', 'runnerStatusCode': 'PENDING',
+ # 'duration': -1, 'location': {'host': None, 'port': -1, 'tlsPort': -1},
+ # 'dataSource': 'w000', 'errorMsg': None}}
+ self._status = self._tasks().task_status(self._id)
+ self._state = self._status['status']['status']
+ if self._state == consts.FAILED_STATE:
+ self._error = self._status['status']['errorMsg']
+ return self._status
+
+ def done(self):
+ """
+ Reports if the query is done: succeeded or failed.
+ """
+ return self._state == consts.FAILED_STATE or self._state == consts.SUCCESS_STATE
+
+ def succeeded(self):
+ """
+ Reports if the query succeeded.
+ """
+ return self._state == consts.SUCCESS_STATE
+
+ def state(self):
+ """
+ Reports the engine-specific query state.
+
+ Updated after each call to status().
+ """
+ return self._state
+
+ def error(self):
+ return self._error
+
+ def error_msg(self):
+ err = self.error()
+ if err is None:
+ return "unknown"
+ if type(err) is str:
+ return err
+ msg = dict_get(err, "error")
+ text = dict_get(err, "errorMessage")
+ if msg is None and text is None:
+ return "unknown"
+ if text is not None:
+ text = text.replace('\\n', '\n')
+ if msg is None:
+ return text
+ if text is None:
+ return msg
+ return msg + ": " + text
+
+ def join(self):
+ if not self.done():
+ self.status()
+ while not self.done():
+ time.sleep(0.5)
+ self.status()
+ return self.succeeded()
+
+ def check_valid(self):
+ if self._id is None:
+ raise ClientError("Operation is invalid on a failed query")
+
+ def wait_done(self):
+ if not self.join():
+ raise DruidError("Query failed: " + self.error_msg())
+
+ def wait(self):
+ self.wait_done()
+ return self.rows()
+
+ def reports(self) -> dict:
+ self.check_valid()
+ if self._reports is None:
+ self.join()
+ self._reports = self._tasks().task_reports(self._id)
+ return self._reports
+
+ def results(self):
+ if self._results is None:
+ rpts = self.reports()
+ self._results = rpts['multiStageQuery']['payload']['results']
+ return self._results
+
+ def schema(self):
+ if self._schema is None:
+ results = self.results()
+ sig = results['signature']
+ sqlTypes = results['sqlTypeNames']
+ size = len(sig)
+ self._schema = []
+ for i in range(size):
+ self._schema.append(ColumnSchema(sig[i]['name'], sqlTypes[i], sig[i]['type']))
+ return self._schema
+
+ def rows(self):
+ if self._rows is None:
+ results = self.results()
+ self._rows = results['results']
+ return self._rows
+
+ def show(self, non_null=False):
+ data = self.rows()
+ if non_null:
+ data = filter_null_cols(data)
+ disp = display.display.table()
+ disp.headers([c.name for c in self.schema()])
+ disp.show(data)
+
+class QueryClient:
+
+ def __init__(self, druid, rest_client=None):
+ self.druid_client = druid
+ self._rest_client = druid.rest_client if rest_client is None else rest_client
+
+ def rest_client(self):
+ return self._rest_client
+
+ def _prepare_query(self, request):
+ if request is None:
+ raise ClientError("No query provided.")
+ if type(request) == str:
+ request = self.sql_request(request)
+ if is_blank(request.sql):
+ raise ClientError("No query provided.")
+ if self.rest_client().trace:
+ print(request.sql)
+ query_obj = request.to_request()
+ return (request, query_obj)
+
+ def sql_query(self, request) -> SqlQueryResult:
+ '''
+ Submit a SQL query with control over the context, parameters and other
+ options. Returns a response with either a detailed error message, or
+ the rows and query ID.
+ '''
+ request, query_obj = self._prepare_query(request)
+ r = self.rest_client().post_only_json(REQ_ROUTER_SQL, query_obj, headers=request.headers)
+ return SqlQueryResult(request, r)
+
+ def sql(self, sql, *args):
+ if len(args) > 0:
+ sql = sql.result_format(*args)
+ resp = self.sql_query(sql)
+ if resp.ok():
+ return resp.rows()
+ raise ClientError(resp.error_msg())
+
+ def explain_sql(self, query):
+ """
+ Run an EXPLAIN PLAN FOR query for the given query.
Review Comment:
```suggestion
Runs an EXPLAIN PLAN FOR query for the given SQL query.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/status.py:
##########
@@ -0,0 +1,99 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time
+
+STATUS_BASE = "/status"
+REQ_STATUS = STATUS_BASE
+REQ_HEALTH = STATUS_BASE + "/health"
+REQ_PROPERTIES = STATUS_BASE + "/properties"
+REQ_IN_CLUSTER = STATUS_BASE + "/selfDiscovered/status"
+
+ROUTER_BASE = '/druid/router/v1'
+REQ_BROKERS = ROUTER_BASE + '/brokers'
+
+class StatusClient:
+ '''
+ Client for status APIs. These APIs are available on all nodes.
+ If used with the router, they report the status of just the router.
+ '''
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ #-------- Common --------
+
+ def status(self):
+ """
+ Returns the Druid version, loaded extensions, memory used, total memory
+ and other useful information about the process.
+
+ GET `/status`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#process-information
+ """
+ return self.client.get_json(REQ_STATUS)
+
+ def is_healthy(self) -> bool:
+ """
+ Returns `True` if the node is healthy, an exception otherwise.
+ Useful for automated health checks.
+
+ GET `/status/health`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#process-information
+ """
+ try:
+ return self.client.get_json(REQ_HEALTH)
+ except Exception:
+ return False
+
+ def wait_until_ready(self):
+ while not self.is_healthy():
+ time.sleep(0.5)
+
+ def properties(self) -> map:
+ """
+ Returns the effective set of Java properties used by the service, including
+ system properties and properties from the `common_runtime.propeties` and
+ `runtime.properties` files.
+
+ GET `/status/properties`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#process-information
+ """
+ return self.client.get_json(REQ_PROPERTIES)
+
+ def in_cluster(self):
+ """
+ Returns `True` if the node is visible wihtin the cluster, `False` if not.
Review Comment:
```suggestion
Returns `True` if the node is visible within the cluster, `False` if not.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/sql.py:
##########
@@ -0,0 +1,690 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time, requests
+from . import consts, display
+from .consts import ROUTER_BASE
+from .util import is_blank, dict_get
+from .error import DruidError, ClientError
+
+REQ_ROUTER_QUERY = ROUTER_BASE
+REQ_ROUTER_SQL = ROUTER_BASE + '/sql'
+REQ_ROUTER_SQL_TASK = REQ_ROUTER_SQL + '/task'
+
+class SqlRequest:
+
+ def __init__(self, query_client, sql):
+ self.query_client = query_client
+ self.sql = sql
+ self.context = None
+ self.params = None
+ self.header = False
+ self.format = consts.SQL_OBJECT
+ self.headers = None
+ self.types = None
+ self.sqlTypes = None
+
+ def with_format(self, result_format):
+ self.format = result_format
+ return self
+
+ def with_headers(self, sqlTypes=False, druidTypes=False):
+ self.headers = True
+ self.types = druidTypes
+ self.sqlTypes = sqlTypes
+ return self
+
+ def with_context(self, context):
+ if self.context is None:
+ self.context = context
+ else:
+ self.context.update(context)
+ return self
+
+ def with_parameters(self, params):
+ '''
+ Set the array of parameters. Parameters must each be a map of 'type'/'value' pairs:
+ {'type': the_type, 'value': the_value}. The type must be a valid SQL type
+ (in upper case). See the consts module for a list.
+ '''
+ if self.params is None:
+ self.params = params
+ else:
+ self.params.update(params)
+ return self
+
+ def add_parameter(self, value):
+ '''
+ Add one parameter value. Infers the type of the parameter from the Python type.
+ '''
+ if value is None:
+ raise ClientError("Druid does not support null parameter values")
+ data_type = None
+ value_type = type(value)
+ if value_type is str:
+ data_type = consts.SQL_VARCHAR_TYPE
+ elif value_type is int:
+ data_type = consts.SQL_BIGINT_TYPE
+ elif value_type is float:
+ data_type = consts.SQL_DOUBLE_TYPE
+ elif value_type is list:
+ data_type = consts.SQL_ARRAY_TYPE
+ else:
+ raise ClientError("Unsupported value type")
+ if self.params is None:
+ self.params = []
+ self.params.append({'type': data_type, 'value': value})
+
+ def response_header(self):
+ self.header = True
+ return self
+
+ def request_headers(self, headers):
+ self.headers = headers
+ return self
+
+ def to_request(self):
+ query_obj = {"query": self.sql}
+ if self.context is not None and len(self.context) > 0:
+ query_obj['context'] = self.context
+ if self.params is not None and len(self.params) > 0:
+ query_obj['parameters'] = self.params
+ if self.header:
+ query_obj['header'] = True
+ if self.result_format is not None:
+ query_obj['resultFormat'] = self.format
+ if self.sqlTypes:
+ query_obj['sqlTypesHeader'] = self.sqlTypes
+ if self.types:
+ query_obj['typesHeader'] = self.types
+ return query_obj
+
+ def result_format(self):
+ return self.format.lower()
+
+ def run(self):
+ return self.query_client.sql_query(self)
+
+def parse_rows(fmt, context, results):
+ if fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ rows = results['results']
+ elif fmt == consts.SQL_ARRAY:
+ rows = results
+ else:
+ return results
+ if not context.get('headers', False):
+ return rows
+ header_size = 1
+ if context.get('sqlTypesHeader', False):
+ header_size += 1
+ if context.get('typesHeader', False):
+ header_size += 1
+ return rows[header_size:]
+
+def label_non_null_cols(results):
+ if results is None or len(results) == 0:
+ return []
+ is_null = {}
+ for key in results[0].keys():
+ is_null[key] = True
+ for row in results:
+ for key, value in row.items():
+ if type(value) == str:
+ if value != '':
+ is_null[key] = False
+ elif type(value) == float:
+ if value != 0.0:
+ is_null[key] = False
+ elif value is not None:
+ is_null[key] = False
+ return is_null
+
+def filter_null_cols(results):
+ '''
+ Filter columns from a Druid result set by removing all null-like
+ columns. A column is considered null if all values for that column
+ are null. A value is null if it is either a JSON null, an empty
+ string, or a numeric 0. All rows are preserved, as is the order
+ of the remaining columns.
+ '''
+ if results is None or len(results) == 0:
+ return results
+ is_null = label_non_null_cols(results)
+ revised = []
+ for row in results:
+ new_row = {}
+ for key, value in row.items():
+ if is_null[key]:
+ continue
+ new_row[key] = value
+ revised.append(new_row)
+ return revised
+
+def parse_object_schema(results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ row = results[0]
+ for k, v in row.items():
+ druid_type = None
+ sql_type = None
+ if type(v) is str:
+ druid_type = consts.DRUID_STRING_TYPE
+ sql_type = consts.SQL_VARCHAR_TYPE
+ elif type(v) is int or type(v) is float:
+ druid_type = consts.DRUID_LONG_TYPE
+ sql_type = consts.SQL_BIGINT_TYPE
+ schema.append(ColumnSchema(k, sql_type, druid_type))
+ return schema
+
+def parse_array_schema(context, results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ has_headers = context.get(consts.HEADERS_KEY, False)
+ if not has_headers:
+ return schema
+ has_sql_types = context.get(consts.SQL_TYPES_HEADERS_KEY, False)
+ has_druid_types = context.get(consts.DRUID_TYPE_HEADERS_KEY, False)
+ size = len(results[0])
+ for i in range(size):
+ druid_type = None
+ if has_druid_types:
+ druid_type = results[1][i]
+ sql_type = None
+ if has_sql_types:
+ sql_type = results[2][i]
+ schema.append(ColumnSchema(results[0][i], sql_type, druid_type))
+ return schema
+
+def parse_schema(fmt, context, results):
+ if fmt == consts.SQL_OBJECT:
+ return parse_object_schema(results)
+ elif fmt == consts.SQL_ARRAY or fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ return parse_array_schema(context, results)
+ else:
+ return []
+
+def is_response_ok(http_response):
+ code = http_response.status_code
+ return code == requests.codes.ok or code == requests.codes.accepted
+
+class ColumnSchema:
+
+ def __init__(self, name, sql_type, druid_type):
+ self.name = name
+ self.sql_type = sql_type
+ self.druid_type = druid_type
+
+ def __str__(self):
+ return "{{name={}, SQL type={}, Druid type={}}}".format(self.name, self.sql_type, self.druid_type)
+
+class SqlQueryResult:
+ """
+ Defines the core protocol for Druid SQL queries.
+ """
+
+ def __init__(self, request, response):
+ self.http_response = response
+ self._json = None
+ self._rows = None
+ self._schema = None
+ self.request = request
+ self._error = None
+ self._id = None
+ if not is_response_ok(response):
+ try:
+ self._error = response.json()
+ except Exception:
+ self._error = response.text
+ if self._error is None or len(self._error) == 0:
+ self._error = "Failed with HTTP status {}".format(response.status_code)
+ try:
+ self._id = self.http_response.headers['X-Druid-SQL-Query-Id']
+ except KeyError:
+ self._error = "Query returned no query ID"
+
+ def result_format(self):
+ return self.request.result_format()
+
+ def ok(self):
+ """
+ Reports if the query succeeded.
+
+ The query rows and schema are available only if ok() returns True.
+ """
+ return is_response_ok(self.http_response)
+
+ def error_msg(self):
+ err = self.error()
+ if err is None:
+ return "unknown"
+ if type(err) is str:
+ return err
+ msg = err.get("error")
+ text = err.get("errorMessage")
+ if msg is None and text is None:
+ return "unknown"
+ if msg is None:
+ return text
+ if text is None:
+ return msg
+ return msg + ": " + text
+
+ def id(self):
+ """
+ Returns the unique identifier for the query.
+ """
+ return self._id
+
+ def non_null(self):
+ if not self.ok():
+ return None
+ if self.result_format() != consts.SQL_OBJECT:
+ return None
+ return filter_null_cols(self.rows())
+
+ def as_array(self):
+ if self.result_format() == consts.SQL_OBJECT:
+ rows = []
+ for obj in self.rows():
+ rows.append([v for v in obj.values()])
+ return rows
+ else:
+ return self.rows()
+
+ def error(self):
+ """
+ If the query fails, returns the error, if any provided by Druid.
+ """
+ if self.ok():
+ return None
+ if self._error is not None:
+ return self._error
+ if self.http_response is None:
+ return { "error": "unknown"}
+ if is_response_ok(self.http_response):
+ return None
+ return {"error": "HTTP {}".format(self.http_response.status_code)}
+
+ def json(self):
+ if not self.ok():
+ return None
+ if self._json is None:
+ self._json = self.http_response.json()
+ return self._json
+
+ def rows(self):
+ """
+ Returns the rows of data for the query.
+
+ Druid supports many data formats. The method makes its best
+ attempt to map the format into an array of rows of some sort.
+ """
+ if self._rows is None:
+ json = self.json()
+ if json is None:
+ return self.http_response.text
+ self._rows = parse_rows(self.result_format(), self.request.context, json)
+ return self._rows
+
+ def schema(self):
+ """
+ Returns the data schema as a list of ColumnSchema objects.
+
+ Druid supports many data formats, not all of them provide
+ schema information. This method makes its best attempt to
+ extract the schema from the query results.
+ """
+ if self._schema is None:
+ self._schema = parse_schema(self.result_format(), self.request.context, self.json())
+ return self._schema
+
+ def show(self, non_null=False):
+ data = None
+ if non_null:
+ data = self.non_null()
+ if data is None:
+ data = self.as_array()
+ if data is None or len(data) == 0:
+ display.display.show_message("Query returned no results")
+ return
+ disp = display.display.table()
+ disp.headers([c.name for c in self.schema()])
+ disp.show(data)
+
+ def show_schema(self):
+ disp = display.display.table()
+ disp.headers(['Name', 'SQL Type', 'Druid Type'])
+ data = []
+ for c in self.schema():
+ data.append([c.name, c.sql_type, c.druid_type])
+ disp.show(data)
+
+class QueryTaskResult:
+
+ def __init__(self, request, response):
+ self._request = request
+ self.http_response = response
+ self._status = None
+ self._results = None
+ self._details = None
+ self._schema = None
+ self._rows = None
+ self._reports = None
+ self._schema = None
+ self._results = None
+ self._error = None
+ self._id = None
+ if not is_response_ok(response):
+ self._state = consts.FAILED_STATE
+ try:
+ self._error = response.json()
+ except Exception:
+ self._error = response.text
+ if self._error is None or len(self._error) == 0:
+ self._error = "Failed with HTTP status {}".format(response.status_code)
+ return
+
+ # Typical response:
+ # {'taskId': '6f7b514a446d4edc9d26a24d4bd03ade_fd8e242b-7d93-431d-b65b-2a512116924c_bjdlojgj',
+ # 'state': 'RUNNING'}
+ self.response_obj = response.json()
+ self._id = self.response_obj['taskId']
+ self._state = self.response_obj['state']
+
+ def ok(self):
+ """
+ Reports if the query succeeded.
+
+ The query rows and schema are available only if ok() returns True.
+ """
+ return self._error is None
+
+ def id(self):
+ return self._id
+
+ def _tasks(self):
+ return self._request.query_client.druid_client.tasks()
+
+ def status(self):
+ """
+ Polls Druid for an update on the query run status.
+ """
+ self.check_valid()
+ # Example:
+ # {'task': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'status': {
+ # 'id': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'groupId': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'type': 'talaria0', 'createdTime': '2022-04-28T23:19:50.331Z',
+ # 'queueInsertionTime': '1970-01-01T00:00:00.000Z',
+ # 'statusCode': 'RUNNING', 'status': 'RUNNING', 'runnerStatusCode': 'PENDING',
+ # 'duration': -1, 'location': {'host': None, 'port': -1, 'tlsPort': -1},
+ # 'dataSource': 'w000', 'errorMsg': None}}
+ self._status = self._tasks().task_status(self._id)
+ self._state = self._status['status']['status']
+ if self._state == consts.FAILED_STATE:
+ self._error = self._status['status']['errorMsg']
+ return self._status
+
+ def done(self):
+ """
+ Reports if the query is done: succeeded or failed.
+ """
+ return self._state == consts.FAILED_STATE or self._state == consts.SUCCESS_STATE
+
+ def succeeded(self):
+ """
+ Reports if the query succeeded.
+ """
+ return self._state == consts.SUCCESS_STATE
+
+ def state(self):
+ """
+ Reports the engine-specific query state.
+
+ Updated after each call to status().
+ """
+ return self._state
+
+ def error(self):
+ return self._error
+
+ def error_msg(self):
+ err = self.error()
+ if err is None:
+ return "unknown"
+ if type(err) is str:
+ return err
+ msg = dict_get(err, "error")
+ text = dict_get(err, "errorMessage")
+ if msg is None and text is None:
+ return "unknown"
+ if text is not None:
+ text = text.replace('\\n', '\n')
+ if msg is None:
+ return text
+ if text is None:
+ return msg
+ return msg + ": " + text
+
+ def join(self):
+ if not self.done():
+ self.status()
+ while not self.done():
+ time.sleep(0.5)
+ self.status()
+ return self.succeeded()
+
+ def check_valid(self):
+ if self._id is None:
+ raise ClientError("Operation is invalid on a failed query")
+
+ def wait_done(self):
+ if not self.join():
+ raise DruidError("Query failed: " + self.error_msg())
+
+ def wait(self):
+ self.wait_done()
+ return self.rows()
+
+ def reports(self) -> dict:
+ self.check_valid()
+ if self._reports is None:
+ self.join()
+ self._reports = self._tasks().task_reports(self._id)
+ return self._reports
+
+ def results(self):
+ if self._results is None:
+ rpts = self.reports()
+ self._results = rpts['multiStageQuery']['payload']['results']
+ return self._results
+
+ def schema(self):
+ if self._schema is None:
+ results = self.results()
+ sig = results['signature']
+ sqlTypes = results['sqlTypeNames']
+ size = len(sig)
+ self._schema = []
+ for i in range(size):
+ self._schema.append(ColumnSchema(sig[i]['name'], sqlTypes[i], sig[i]['type']))
+ return self._schema
+
+ def rows(self):
+ if self._rows is None:
+ results = self.results()
+ self._rows = results['results']
+ return self._rows
+
+ def show(self, non_null=False):
+ data = self.rows()
+ if non_null:
+ data = filter_null_cols(data)
+ disp = display.display.table()
+ disp.headers([c.name for c in self.schema()])
+ disp.show(data)
+
+class QueryClient:
+
+ def __init__(self, druid, rest_client=None):
+ self.druid_client = druid
+ self._rest_client = druid.rest_client if rest_client is None else rest_client
+
+ def rest_client(self):
+ return self._rest_client
+
+ def _prepare_query(self, request):
+ if request is None:
+ raise ClientError("No query provided.")
+ if type(request) == str:
+ request = self.sql_request(request)
+ if is_blank(request.sql):
+ raise ClientError("No query provided.")
+ if self.rest_client().trace:
+ print(request.sql)
+ query_obj = request.to_request()
+ return (request, query_obj)
+
+ def sql_query(self, request) -> SqlQueryResult:
+ '''
+ Submit a SQL query with control over the context, parameters and other
+ options. Returns a response with either a detailed error message, or
+ the rows and query ID.
Review Comment:
Are the rows and query ID returned for a successful response? If so, maybe can clarify with something like "If successful, returns a response with the rows and query ID, else returns a detailed error message."
##########
examples/quickstart/jupyter-notebooks/druidapi/status.py:
##########
@@ -0,0 +1,99 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time
+
+STATUS_BASE = "/status"
+REQ_STATUS = STATUS_BASE
+REQ_HEALTH = STATUS_BASE + "/health"
+REQ_PROPERTIES = STATUS_BASE + "/properties"
+REQ_IN_CLUSTER = STATUS_BASE + "/selfDiscovered/status"
+
+ROUTER_BASE = '/druid/router/v1'
+REQ_BROKERS = ROUTER_BASE + '/brokers'
+
+class StatusClient:
+ '''
+ Client for status APIs. These APIs are available on all nodes.
+ If used with the router, they report the status of just the router.
+ '''
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ #-------- Common --------
+
+ def status(self):
+ """
+ Returns the Druid version, loaded extensions, memory used, total memory
+ and other useful information about the process.
+
+ GET `/status`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#process-information
+ """
+ return self.client.get_json(REQ_STATUS)
+
+ def is_healthy(self) -> bool:
+ """
+ Returns `True` if the node is healthy, an exception otherwise.
+ Useful for automated health checks.
+
+ GET `/status/health`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#process-information
+ """
+ try:
+ return self.client.get_json(REQ_HEALTH)
+ except Exception:
+ return False
+
+ def wait_until_ready(self):
+ while not self.is_healthy():
+ time.sleep(0.5)
+
+ def properties(self) -> map:
+ """
+ Returns the effective set of Java properties used by the service, including
+ system properties and properties from the `common_runtime.propeties` and
+ `runtime.properties` files.
+
+ GET `/status/properties`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#process-information
+ """
+ return self.client.get_json(REQ_PROPERTIES)
+
+ def in_cluster(self):
+ """
+ Returns `True` if the node is visible wihtin the cluster, `False` if not.
+ (That is, returns the value of the `{"selfDiscovered": true/false}`
Review Comment:
```suggestion
That is, returns the value of the `{"selfDiscovered": true/false}`
```
##########
examples/quickstart/jupyter-notebooks/druidapi/tasks.py:
##########
@@ -0,0 +1,178 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .consts import OVERLORD_BASE
+
+# Tasks
+REQ_TASKS = OVERLORD_BASE + '/tasks'
+REQ_POST_TASK = OVERLORD_BASE + '/task'
+REQ_GET_TASK = REQ_POST_TASK + '/{}'
+REQ_TASK_STATUS = REQ_GET_TASK + '/status'
+REQ_TASK_REPORTS = REQ_GET_TASK + '/reports'
+REQ_END_TASK = REQ_GET_TASK
+REQ_END_DS_TASKS = REQ_END_TASK + '/shutdownAllTasks'
+
+class TaskClient:
+ """
+ Client for task-related APIs. The APIs connect through the Router to
+ the Overlord.
+ """
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ def tasks(self, state=None, table=None, type=None, max=None, created_time_interval=None):
+ '''
+ Retrieve list of tasks.
+
+ Parameters
+ ----------
+ state : str, default = None
+ Filter list of tasks by task state. Valid options are "running",
+ "complete", "waiting", and "pending". Constants are defined for
+ each of these in the `consts` file.
+ table : str, default = None
+ Return tasks filtered by Druid table (datasource).
+ created_time_interval : str, Default = None
+ Return tasks created within the specified interval.
+ max : int, default = None
+ Maximum number of "complete" tasks to return. Only applies when state is set to "complete".
+ type : str, default = None
+ filter tasks by task type.
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/tasks`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ params = {}
+ if state is not None:
+ params['state'] = state
+ if table is not None:
+ params['datasource'] = table
+ if type is not None:
+ params['type'] = type
+ if max is not None:
+ params['max'] = max
+ if created_time_interval is not None:
+ params['createdTimeInterval'] = created_time_interval
+ return self.client.get_json(REQ_TASKS, params=params)
+
+ def task(self, task_id):
+ """
+ Retrieve the "payload" of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ """
+ return self.client.get_json(REQ_GET_TASK, args=[task_id])
+
+ def task_status(self, task_id):
+ '''
+ Retrieve the status of a task.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
+
+ Reference
+ ---------
+ `GET /druid/indexer/v1/task/{taskId}/status`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#get-15
+ '''
+ return self.client.get_json(REQ_TASK_STATUS, args=[task_id])
+
+ def task_reports(self, task_id):
+ '''
+ Retrieve a task completion report for a task.
+ Only works for completed tasks.
+
+ Parameters
+ ----------
+ task_id : str
+ The id of the task to retrieve
Review Comment:
```suggestion
The ID of the task to retrieve.
```
##########
examples/quickstart/jupyter-notebooks/druidapi/status.py:
##########
@@ -0,0 +1,99 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time
+
+STATUS_BASE = "/status"
+REQ_STATUS = STATUS_BASE
+REQ_HEALTH = STATUS_BASE + "/health"
+REQ_PROPERTIES = STATUS_BASE + "/properties"
+REQ_IN_CLUSTER = STATUS_BASE + "/selfDiscovered/status"
+
+ROUTER_BASE = '/druid/router/v1'
+REQ_BROKERS = ROUTER_BASE + '/brokers'
+
+class StatusClient:
+ '''
+ Client for status APIs. These APIs are available on all nodes.
+ If used with the router, they report the status of just the router.
+ '''
+
+ def __init__(self, rest_client):
+ self.client = rest_client
+
+ #-------- Common --------
+
+ def status(self):
+ """
+ Returns the Druid version, loaded extensions, memory used, total memory
+ and other useful information about the process.
+
+ GET `/status`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#process-information
+ """
+ return self.client.get_json(REQ_STATUS)
+
+ def is_healthy(self) -> bool:
+ """
+ Returns `True` if the node is healthy, an exception otherwise.
+ Useful for automated health checks.
+
+ GET `/status/health`
+
+ See https://druid.apache.org/docs/latest/operations/api-reference.html#process-information
+ """
+ try:
+ return self.client.get_json(REQ_HEALTH)
+ except Exception:
+ return False
+
+ def wait_until_ready(self):
+ while not self.is_healthy():
+ time.sleep(0.5)
+
+ def properties(self) -> map:
+ """
+ Returns the effective set of Java properties used by the service, including
+ system properties and properties from the `common_runtime.propeties` and
Review Comment:
```suggestion
system properties and properties from the `common_runtime.properties` and
```
##########
examples/quickstart/jupyter-notebooks/druidapi/sql.py:
##########
@@ -0,0 +1,690 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time, requests
+from . import consts, display
+from .consts import ROUTER_BASE
+from .util import is_blank, dict_get
+from .error import DruidError, ClientError
+
+REQ_ROUTER_QUERY = ROUTER_BASE
+REQ_ROUTER_SQL = ROUTER_BASE + '/sql'
+REQ_ROUTER_SQL_TASK = REQ_ROUTER_SQL + '/task'
+
+class SqlRequest:
+
+ def __init__(self, query_client, sql):
+ self.query_client = query_client
+ self.sql = sql
+ self.context = None
+ self.params = None
+ self.header = False
+ self.format = consts.SQL_OBJECT
+ self.headers = None
+ self.types = None
+ self.sqlTypes = None
+
+ def with_format(self, result_format):
+ self.format = result_format
+ return self
+
+ def with_headers(self, sqlTypes=False, druidTypes=False):
+ self.headers = True
+ self.types = druidTypes
+ self.sqlTypes = sqlTypes
+ return self
+
+ def with_context(self, context):
+ if self.context is None:
+ self.context = context
+ else:
+ self.context.update(context)
+ return self
+
+ def with_parameters(self, params):
+ '''
+ Set the array of parameters. Parameters must each be a map of 'type'/'value' pairs:
+ {'type': the_type, 'value': the_value}. The type must be a valid SQL type
+ (in upper case). See the consts module for a list.
+ '''
+ if self.params is None:
+ self.params = params
+ else:
+ self.params.update(params)
+ return self
+
+ def add_parameter(self, value):
+ '''
+ Add one parameter value. Infers the type of the parameter from the Python type.
+ '''
+ if value is None:
+ raise ClientError("Druid does not support null parameter values")
+ data_type = None
+ value_type = type(value)
+ if value_type is str:
+ data_type = consts.SQL_VARCHAR_TYPE
+ elif value_type is int:
+ data_type = consts.SQL_BIGINT_TYPE
+ elif value_type is float:
+ data_type = consts.SQL_DOUBLE_TYPE
+ elif value_type is list:
+ data_type = consts.SQL_ARRAY_TYPE
+ else:
+ raise ClientError("Unsupported value type")
+ if self.params is None:
+ self.params = []
+ self.params.append({'type': data_type, 'value': value})
+
+ def response_header(self):
+ self.header = True
+ return self
+
+ def request_headers(self, headers):
+ self.headers = headers
+ return self
+
+ def to_request(self):
+ query_obj = {"query": self.sql}
+ if self.context is not None and len(self.context) > 0:
+ query_obj['context'] = self.context
+ if self.params is not None and len(self.params) > 0:
+ query_obj['parameters'] = self.params
+ if self.header:
+ query_obj['header'] = True
+ if self.result_format is not None:
+ query_obj['resultFormat'] = self.format
+ if self.sqlTypes:
+ query_obj['sqlTypesHeader'] = self.sqlTypes
+ if self.types:
+ query_obj['typesHeader'] = self.types
+ return query_obj
+
+ def result_format(self):
+ return self.format.lower()
+
+ def run(self):
+ return self.query_client.sql_query(self)
+
+def parse_rows(fmt, context, results):
+ if fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ rows = results['results']
+ elif fmt == consts.SQL_ARRAY:
+ rows = results
+ else:
+ return results
+ if not context.get('headers', False):
+ return rows
+ header_size = 1
+ if context.get('sqlTypesHeader', False):
+ header_size += 1
+ if context.get('typesHeader', False):
+ header_size += 1
+ return rows[header_size:]
+
+def label_non_null_cols(results):
+ if results is None or len(results) == 0:
+ return []
+ is_null = {}
+ for key in results[0].keys():
+ is_null[key] = True
+ for row in results:
+ for key, value in row.items():
+ if type(value) == str:
+ if value != '':
+ is_null[key] = False
+ elif type(value) == float:
+ if value != 0.0:
+ is_null[key] = False
+ elif value is not None:
+ is_null[key] = False
+ return is_null
+
+def filter_null_cols(results):
+ '''
+ Filter columns from a Druid result set by removing all null-like
+ columns. A column is considered null if all values for that column
+ are null. A value is null if it is either a JSON null, an empty
+ string, or a numeric 0. All rows are preserved, as is the order
+ of the remaining columns.
+ '''
+ if results is None or len(results) == 0:
+ return results
+ is_null = label_non_null_cols(results)
+ revised = []
+ for row in results:
+ new_row = {}
+ for key, value in row.items():
+ if is_null[key]:
+ continue
+ new_row[key] = value
+ revised.append(new_row)
+ return revised
+
+def parse_object_schema(results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ row = results[0]
+ for k, v in row.items():
+ druid_type = None
+ sql_type = None
+ if type(v) is str:
+ druid_type = consts.DRUID_STRING_TYPE
+ sql_type = consts.SQL_VARCHAR_TYPE
+ elif type(v) is int or type(v) is float:
+ druid_type = consts.DRUID_LONG_TYPE
+ sql_type = consts.SQL_BIGINT_TYPE
+ schema.append(ColumnSchema(k, sql_type, druid_type))
+ return schema
+
+def parse_array_schema(context, results):
+ schema = []
+ if len(results) == 0:
+ return schema
+ has_headers = context.get(consts.HEADERS_KEY, False)
+ if not has_headers:
+ return schema
+ has_sql_types = context.get(consts.SQL_TYPES_HEADERS_KEY, False)
+ has_druid_types = context.get(consts.DRUID_TYPE_HEADERS_KEY, False)
+ size = len(results[0])
+ for i in range(size):
+ druid_type = None
+ if has_druid_types:
+ druid_type = results[1][i]
+ sql_type = None
+ if has_sql_types:
+ sql_type = results[2][i]
+ schema.append(ColumnSchema(results[0][i], sql_type, druid_type))
+ return schema
+
+def parse_schema(fmt, context, results):
+ if fmt == consts.SQL_OBJECT:
+ return parse_object_schema(results)
+ elif fmt == consts.SQL_ARRAY or fmt == consts.SQL_ARRAY_WITH_TRAILER:
+ return parse_array_schema(context, results)
+ else:
+ return []
+
+def is_response_ok(http_response):
+ code = http_response.status_code
+ return code == requests.codes.ok or code == requests.codes.accepted
+
+class ColumnSchema:
+
+ def __init__(self, name, sql_type, druid_type):
+ self.name = name
+ self.sql_type = sql_type
+ self.druid_type = druid_type
+
+ def __str__(self):
+ return "{{name={}, SQL type={}, Druid type={}}}".format(self.name, self.sql_type, self.druid_type)
+
+class SqlQueryResult:
+ """
+ Defines the core protocol for Druid SQL queries.
+ """
+
+ def __init__(self, request, response):
+ self.http_response = response
+ self._json = None
+ self._rows = None
+ self._schema = None
+ self.request = request
+ self._error = None
+ self._id = None
+ if not is_response_ok(response):
+ try:
+ self._error = response.json()
+ except Exception:
+ self._error = response.text
+ if self._error is None or len(self._error) == 0:
+ self._error = "Failed with HTTP status {}".format(response.status_code)
+ try:
+ self._id = self.http_response.headers['X-Druid-SQL-Query-Id']
+ except KeyError:
+ self._error = "Query returned no query ID"
+
+ def result_format(self):
+ return self.request.result_format()
+
+ def ok(self):
+ """
+ Reports if the query succeeded.
+
+ The query rows and schema are available only if ok() returns True.
+ """
+ return is_response_ok(self.http_response)
+
+ def error_msg(self):
+ err = self.error()
+ if err is None:
+ return "unknown"
+ if type(err) is str:
+ return err
+ msg = err.get("error")
+ text = err.get("errorMessage")
+ if msg is None and text is None:
+ return "unknown"
+ if msg is None:
+ return text
+ if text is None:
+ return msg
+ return msg + ": " + text
+
+ def id(self):
+ """
+ Returns the unique identifier for the query.
+ """
+ return self._id
+
+ def non_null(self):
+ if not self.ok():
+ return None
+ if self.result_format() != consts.SQL_OBJECT:
+ return None
+ return filter_null_cols(self.rows())
+
+ def as_array(self):
+ if self.result_format() == consts.SQL_OBJECT:
+ rows = []
+ for obj in self.rows():
+ rows.append([v for v in obj.values()])
+ return rows
+ else:
+ return self.rows()
+
+ def error(self):
+ """
+ If the query fails, returns the error, if any provided by Druid.
+ """
+ if self.ok():
+ return None
+ if self._error is not None:
+ return self._error
+ if self.http_response is None:
+ return { "error": "unknown"}
+ if is_response_ok(self.http_response):
+ return None
+ return {"error": "HTTP {}".format(self.http_response.status_code)}
+
+ def json(self):
+ if not self.ok():
+ return None
+ if self._json is None:
+ self._json = self.http_response.json()
+ return self._json
+
+ def rows(self):
+ """
+ Returns the rows of data for the query.
+
+ Druid supports many data formats. The method makes its best
+ attempt to map the format into an array of rows of some sort.
+ """
+ if self._rows is None:
+ json = self.json()
+ if json is None:
+ return self.http_response.text
+ self._rows = parse_rows(self.result_format(), self.request.context, json)
+ return self._rows
+
+ def schema(self):
+ """
+ Returns the data schema as a list of ColumnSchema objects.
+
+ Druid supports many data formats, not all of them provide
+ schema information. This method makes its best attempt to
+ extract the schema from the query results.
+ """
+ if self._schema is None:
+ self._schema = parse_schema(self.result_format(), self.request.context, self.json())
+ return self._schema
+
+ def show(self, non_null=False):
+ data = None
+ if non_null:
+ data = self.non_null()
+ if data is None:
+ data = self.as_array()
+ if data is None or len(data) == 0:
+ display.display.show_message("Query returned no results")
+ return
+ disp = display.display.table()
+ disp.headers([c.name for c in self.schema()])
+ disp.show(data)
+
+ def show_schema(self):
+ disp = display.display.table()
+ disp.headers(['Name', 'SQL Type', 'Druid Type'])
+ data = []
+ for c in self.schema():
+ data.append([c.name, c.sql_type, c.druid_type])
+ disp.show(data)
+
+class QueryTaskResult:
+
+ def __init__(self, request, response):
+ self._request = request
+ self.http_response = response
+ self._status = None
+ self._results = None
+ self._details = None
+ self._schema = None
+ self._rows = None
+ self._reports = None
+ self._schema = None
+ self._results = None
+ self._error = None
+ self._id = None
+ if not is_response_ok(response):
+ self._state = consts.FAILED_STATE
+ try:
+ self._error = response.json()
+ except Exception:
+ self._error = response.text
+ if self._error is None or len(self._error) == 0:
+ self._error = "Failed with HTTP status {}".format(response.status_code)
+ return
+
+ # Typical response:
+ # {'taskId': '6f7b514a446d4edc9d26a24d4bd03ade_fd8e242b-7d93-431d-b65b-2a512116924c_bjdlojgj',
+ # 'state': 'RUNNING'}
+ self.response_obj = response.json()
+ self._id = self.response_obj['taskId']
+ self._state = self.response_obj['state']
+
+ def ok(self):
+ """
+ Reports if the query succeeded.
+
+ The query rows and schema are available only if ok() returns True.
+ """
+ return self._error is None
+
+ def id(self):
+ return self._id
+
+ def _tasks(self):
+ return self._request.query_client.druid_client.tasks()
+
+ def status(self):
+ """
+ Polls Druid for an update on the query run status.
+ """
+ self.check_valid()
+ # Example:
+ # {'task': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'status': {
+ # 'id': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'groupId': 'talaria-sql-w000-b373b68d-2675-4035-b4d2-7a9228edead6',
+ # 'type': 'talaria0', 'createdTime': '2022-04-28T23:19:50.331Z',
+ # 'queueInsertionTime': '1970-01-01T00:00:00.000Z',
+ # 'statusCode': 'RUNNING', 'status': 'RUNNING', 'runnerStatusCode': 'PENDING',
+ # 'duration': -1, 'location': {'host': None, 'port': -1, 'tlsPort': -1},
+ # 'dataSource': 'w000', 'errorMsg': None}}
+ self._status = self._tasks().task_status(self._id)
+ self._state = self._status['status']['status']
+ if self._state == consts.FAILED_STATE:
+ self._error = self._status['status']['errorMsg']
+ return self._status
+
+ def done(self):
+ """
+ Reports if the query is done: succeeded or failed.
+ """
+ return self._state == consts.FAILED_STATE or self._state == consts.SUCCESS_STATE
+
+ def succeeded(self):
+ """
+ Reports if the query succeeded.
+ """
+ return self._state == consts.SUCCESS_STATE
+
+ def state(self):
+ """
+ Reports the engine-specific query state.
+
+ Updated after each call to status().
+ """
+ return self._state
+
+ def error(self):
+ return self._error
+
+ def error_msg(self):
+ err = self.error()
+ if err is None:
+ return "unknown"
+ if type(err) is str:
+ return err
+ msg = dict_get(err, "error")
+ text = dict_get(err, "errorMessage")
+ if msg is None and text is None:
+ return "unknown"
+ if text is not None:
+ text = text.replace('\\n', '\n')
+ if msg is None:
+ return text
+ if text is None:
+ return msg
+ return msg + ": " + text
+
+ def join(self):
+ if not self.done():
+ self.status()
+ while not self.done():
+ time.sleep(0.5)
+ self.status()
+ return self.succeeded()
+
+ def check_valid(self):
+ if self._id is None:
+ raise ClientError("Operation is invalid on a failed query")
+
+ def wait_done(self):
+ if not self.join():
+ raise DruidError("Query failed: " + self.error_msg())
+
+ def wait(self):
+ self.wait_done()
+ return self.rows()
+
+ def reports(self) -> dict:
+ self.check_valid()
+ if self._reports is None:
+ self.join()
+ self._reports = self._tasks().task_reports(self._id)
+ return self._reports
+
+ def results(self):
+ if self._results is None:
+ rpts = self.reports()
+ self._results = rpts['multiStageQuery']['payload']['results']
+ return self._results
+
+ def schema(self):
+ if self._schema is None:
+ results = self.results()
+ sig = results['signature']
+ sqlTypes = results['sqlTypeNames']
+ size = len(sig)
+ self._schema = []
+ for i in range(size):
+ self._schema.append(ColumnSchema(sig[i]['name'], sqlTypes[i], sig[i]['type']))
+ return self._schema
+
+ def rows(self):
+ if self._rows is None:
+ results = self.results()
+ self._rows = results['results']
+ return self._rows
+
+ def show(self, non_null=False):
+ data = self.rows()
+ if non_null:
+ data = filter_null_cols(data)
+ disp = display.display.table()
+ disp.headers([c.name for c in self.schema()])
+ disp.show(data)
+
+class QueryClient:
+
+ def __init__(self, druid, rest_client=None):
+ self.druid_client = druid
+ self._rest_client = druid.rest_client if rest_client is None else rest_client
+
+ def rest_client(self):
+ return self._rest_client
+
+ def _prepare_query(self, request):
+ if request is None:
+ raise ClientError("No query provided.")
+ if type(request) == str:
+ request = self.sql_request(request)
+ if is_blank(request.sql):
+ raise ClientError("No query provided.")
+ if self.rest_client().trace:
+ print(request.sql)
+ query_obj = request.to_request()
+ return (request, query_obj)
+
+ def sql_query(self, request) -> SqlQueryResult:
+ '''
+ Submit a SQL query with control over the context, parameters and other
+ options. Returns a response with either a detailed error message, or
+ the rows and query ID.
Review Comment:
```suggestion
Submits a SQL query with control over the context, parameters, and other
options. Returns a response with either a detailed error message, or
the rows and query ID.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org