Merge pull request #206 from mozilla/dev

Fixes for previous changes

active_data/actions/__init__.py

@@ -52,7 +52,7 @@ def send_error(active_data_timer, body, e):
     return Response(value2json(e).encode("utf8"), status=status)
 
 
-def test_mode_wait(query):
+def test_mode_wait(query, please_stop):
     """
     WAIT FOR METADATA TO ARRIVE ON INDEX
     :param query: dict() OF REQUEST BODY
@@ -76,7 +76,7 @@ def test_mode_wait(query):
         ):
             metadata_manager = find_container(alias, after=after).namespace
 
-            timeout = Till(seconds=MINUTE.seconds)
+            timeout = Till(seconds=MINUTE.seconds) | please_stop
             while not timeout:
                 # GET FRESH VERSIONS
                 cols = metadata_manager.get_columns(

active_data/actions/query.py

@@ -20,7 +20,7 @@
 from mo_future import binary_type
 from mo_json import json2value, value2json
 from mo_logs import Except, Log
-from mo_threads.threads import register_thread
+from mo_threads.threads import register_thread, MAIN_THREAD
 from mo_times.timer import Timer
 from pyLibrary.env.flask_wrappers import cors_wrapper
 
@@ -53,7 +53,7 @@ def jx_query(path):
                 data = json2value(text)
                 record_request(flask.request, data, None, None)
                 if data.meta.testing:
-                    test_mode_wait(data)
+                    test_mode_wait(data, MAIN_THREAD.please_stop)
 
             find_table_timer = Timer("find container", verbose=DEBUG)
             with find_table_timer:

active_data/actions/sql.py

@@ -23,7 +23,7 @@
 from mo_logs import Log
 from mo_logs.exceptions import Except
 from mo_testing.fuzzytestcase import assertAlmostEqual
-from mo_threads.threads import register_thread
+from mo_threads.threads import register_thread, MAIN_THREAD
 from mo_times.timer import Timer
 from pyLibrary.env.flask_wrappers import cors_wrapper
 
@@ -61,7 +61,7 @@ def sql_query(path):
                 jx_query = parse_sql(data.sql)
                 if jx_query['from'] != None:
                     if data.meta.testing:
-                        test_mode_wait(jx_query)
+                        test_mode_wait(jx_query, MAIN_THREAD.please_stop)
                     frum = find_container(jx_query['from'], after=None)
                 else:
                     frum = None

requirements.txt

@@ -6,4 +6,4 @@ requests==2.18.4
 moz-sql-parser
 boto==2.48.0
 beautifulsoup4==4.6.0
-cryptography==2.1.4
+

tests/test_jx/test_schema_merging.py

@@ -301,6 +301,41 @@ def test_select2(self):
         }
         self.utils.execute_tests(test)
 
+    @skip("complicated where clause needs support")
+    def test_where(self):
+        test = {
+            "data": [
+                {"k": 1, "a": "b"},
+                {"k": 2, "a": {"b": 1}},
+                {"k": 3, "a": {}},
+                {"k": 4, "a": [{"b": 1}, {"b": 2}]},  # TEST THAT INNER CAN BE MAPPED TO NESTED
+                {"k": 5, "a": {"b": 4}},  # TEST THAT INNER IS MAPPED TO NESTED, AFTER SEEING NESTED
+                {"k": 6, "a": 3},
+                {"k": 7, }
+            ],
+            "query": {
+                "from": TEST_TABLE + ".a",
+                "select": ["k"],
+                "where": {"eq": {"a.b": 1}}
+            },
+            "expecting_list": {
+                "meta": {"format": "list"},
+                "data": [{"k": 2}, {"k": 4}]
+            },
+            "expecting_table": {
+                "meta": {"format": "table"},
+                "header": ["k"],
+                "data": [[2], [4]]
+            },
+            "expecting_cube": {
+                "meta": {"format": "cube"},
+                "data": {
+                    "k": [2, 4]
+                }
+            }
+        }
+        self.utils.execute_tests(test)
+
     # @skip("schema merging not working")
     def test_sum(self):
         test = {
@@ -360,18 +395,7 @@ def test_edge(self):
                     {"b": 4, "v": 5},
                     {"v": 14}
                 ]
-            },
-            # "expecting_table": {
-            #     "meta": {"format": "table"},
-            #     "header": ["a.b"],
-            #     "data": [[8]]
-            # },
-            # "expecting_cube": {
-            #     "meta": {"format": "cube"},
-            #     "data": {
-            #         "a.b": 8
-            #     }
-            # }
+            }
         }
         self.utils.execute_tests(test)
 

vendor/jx_base/__init__.py

@@ -11,13 +11,15 @@
 
 from uuid import uuid4
 
+from mo_json.typed_encoder import EXISTS_TYPE
+
 from jx_base.expressions import jx_expression
 from jx_python.expressions import Literal, Python
 from mo_dots import coalesce, listwrap, wrap
 from mo_dots.datas import register_data
 from mo_dots.lists import last
 from mo_future import is_text, text
-from mo_json import value2json, true, false, null
+from mo_json import value2json, true, false, null, EXISTS
 from mo_logs import Log
 from mo_logs.strings import expand_template, quote
 
@@ -115,8 +117,7 @@ def _constraint(row, rownum, rows):
             "constraint\\n{" + "{code}}\\nnot satisfied {" + "{expect}}\\n{" + "{value|indent}}",
             code={{constraint_expr|quote}}, 
             expect={{constraint}}, 
-            value=row,
-            cause=e
+            value=row
         )
 
     def __init__(self, **kwargs):
@@ -238,6 +239,11 @@ def __str__(self):
             {"not": {"find": {"es_column": "null"}}},
             {"not": {"eq": {"es_column": "string"}}},
             {"not": {"eq": {"es_type": "object", "jx_type": "exists"}}},
+            {
+                "when": {"suffix": {"es_column": "." + EXISTS_TYPE}},
+                "then": {"eq": {"jx_type": EXISTS}},
+                "else": True
+            },
             {"eq": [{"last": "nested_path"}, {"literal": "."}]},
             {
                 "when": {"eq": [{"literal": ".~N~"}, {"right": {"es_column": 4}}]},

vendor/jx_elasticsearch/elasticsearch.py

@@ -11,7 +11,6 @@
 
 import ast
 import re
-from collections import namedtuple
 from copy import deepcopy
 
 from jx_base import Column
@@ -21,6 +20,7 @@
 from mo_files import File, mimetype
 from mo_files.url import URL
 from mo_future import binary_type, generator_types, is_binary, is_text, items, text
+from mo_http import http
 from mo_json import BOOLEAN, EXISTS, NESTED, NUMBER, OBJECT, STRING, json2value, value2json
 from mo_json.typed_encoder import BOOLEAN_TYPE, EXISTS_TYPE, NESTED_TYPE, NUMBER_TYPE, STRING_TYPE, TYPE_PREFIX, \
     json_type_to_inserter_type
@@ -30,8 +30,7 @@
 from mo_math import is_integer, is_number
 from mo_math.randoms import Random
 from mo_threads import Lock, ThreadedQueue, Till, THREAD_STOP, Thread, MAIN_THREAD
-from mo_times import Date, Timer, HOUR, dates, Duration
-from mo_http import http
+from mo_times import Date, Timer, HOUR, Duration
 
 DEBUG = True
 DEBUG_METADATA_UPDATE = False
@@ -1284,9 +1283,9 @@ def parse_properties(parent_index_name, parent_name, nested_path, esProperties):
             nested_path=nested_path
         ))
 
-    for name, property in esProperties.items():
+    for short_name, property in esProperties.items():
         index_name = parent_index_name
-        column_name = concat_field(parent_name, name)
+        column_name = concat_field(parent_name, short_name)
         jx_name = column_name
 
         if property.type == "nested" and property.properties:
@@ -1325,7 +1324,7 @@ def parse_properties(parent_index_name, parent_name, nested_path, esProperties):
         if not property.type:
             continue
 
-        cardinality = 0 if not (property.store or property.enabled) and name != '_id' else None
+        cardinality = 0 if not (property.store or property.enabled) and short_name != '_id' else None
 
         if property.fields:
             child_columns = parse_properties(index_name, column_name, nested_path, property.fields)
@@ -1334,7 +1333,18 @@ def parse_properties(parent_index_name, parent_name, nested_path, esProperties):
                     cc.cardinality = None
             columns.extend(child_columns)
 
-        if property.type in es_type_to_json_type.keys():
+        if short_name == EXISTS_TYPE:
+            columns.append(Column(
+                name=jx_name,
+                es_index=index_name,
+                es_column=column_name,
+                es_type=property.type,
+                jx_type=EXISTS,
+                cardinality=cardinality,
+                last_updated=Date.now(),
+                nested_path=nested_path
+            ))
+        elif property.type in es_type_to_json_type.keys():
             columns.append(Column(
                 name=jx_name,
                 es_index=index_name,
@@ -1345,7 +1355,7 @@ def parse_properties(parent_index_name, parent_name, nested_path, esProperties):
                 last_updated=Date.now(),
                 nested_path=nested_path
             ))
-            if property.index_name and name != property.index_name:
+            if property.index_name and short_name != property.index_name:
                 columns.append(Column(
                     name=jx_name,
                     es_index=index_name,