diff --git a/.gitignore b/.gitignore
index bd100911..b66c2e03 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,3 +1,6 @@
+*.DS_Store
+*.pyc
+.ipynb_checkpoints
.idea
__pycache__/
docs/.DS_Store
diff --git a/README.md b/README.md
index d4d3913f..257685ff 100644
--- a/README.md
+++ b/README.md
@@ -24,21 +24,26 @@ a language created at Google earlier.
## Why?
Logica is for engineers, data scientists and other specialists who want to use
-logic programming syntax when writing queries and pipelines to run on
-[BigQuery](https://cloud.google.com/bigquery).
+logic programming syntax when writing queries and pipelines for databases and datawarehouses.
+Logica programs run on
+[BigQuery](https://cloud.google.com/bigquery), [Postgres](https://postgresql.org) and [SQLite](https://www.sqlite.org/).
-Logica compiles to StandardSQL and gives you access to the power of BigQuery
-engine with the convenience of logic programming syntax. This is useful because
-BigQuery is magnitudes more powerful than state of the art native
-logic programming engines.
+Logica compiles to SQL and gives you access to the power of SQL ecosystem
+with the convenience of logic programming syntax.
+
+This is useful because
+SQL enginers are magnitudes more powerful than state of the art native
+logic programming engines. For example, BigQuery is a distributed datawarehouse and thus logic programs written
+in Logica can be easily parallelized onto thousands of servers. Postgres and SQLite are among most popular databases, they are
+capable of processing substantial volumes of data right on your machine.
We encourage you to try Logica, especially if
* you already use logic programming and need more computational power, **or**
-* you use SQL, but feel unsatisfied about its readability, **or**
+* you already have data in BigQuery, PostgreSQL or SQLite, **or**
* you want to learn logic programming and apply it to processing of Big Data.
-In the future we plan to support more SQL dialects and engines.
+Support for more SQL dialects and engines is coming in the future.
## I have not heard of logic programming. What is it?
diff --git a/colab_logica.py b/colab_logica.py
index cfe4b34e..8efe885a 100755
--- a/colab_logica.py
+++ b/colab_logica.py
@@ -16,13 +16,21 @@
"""Library for using Logica in CoLab."""
+from decimal import Decimal
+import getpass
+import json
+import re
+
from .common import color
from .common import concertina_lib
+from .common import psql_logica
from .compiler import functors
from .compiler import rule_translate
from .compiler import universe
+from .type_inference.research import infer
+
import IPython
from IPython.core.magic import register_cell_magic
@@ -70,6 +78,13 @@
PREAMBLE = None
+if hasattr(concertina_lib, 'graphviz'):
+ DISPLAY_MODE = 'colab'
+else:
+ DISPLAY_MODE = 'colab-text'
+
+DEFAULT_ENGINE = 'bigquery'
+
def SetPreamble(preamble):
global PREAMBLE
@@ -83,6 +98,12 @@ def SetDbConnection(connection):
global DB_CONNECTION
DB_CONNECTION = connection
+def ConnectToPostgres(mode='interactive'):
+ connection = psql_logica.ConnectToPostgres(mode)
+ SetDbConnection(connection)
+ global DEFAULT_ENGINE
+ DEFAULT_ENGINE = 'psql'
+
def EnsureAuthenticatedUser():
global USER_AUTHENTICATED
global PROJECT
@@ -143,9 +164,14 @@ def RunSQL(sql, engine, connection=None, is_final=False):
return client.query(sql).to_dataframe()
elif engine == 'psql':
if is_final:
- return pandas.read_sql(sql, connection)
+ cursor = psql_logica.PostgresExecute(sql, connection)
+ rows = cursor.fetchall()
+ df = pandas.DataFrame(
+ rows, columns=[d[0] for d in cursor.description])
+ df = df.applymap(psql_logica.DigestPsqlType)
+ return df
else:
- return connection.execute(sql)
+ psql_logica.PostgresExecute(sql, connection)
elif engine == 'sqlite':
try:
if is_final:
@@ -164,6 +190,15 @@ def RunSQL(sql, engine, connection=None, is_final=False):
'for now.')
+def Ingress(table_name, csv_file_name):
+ with open(csv_file_name) as csv_data_io:
+ cursor = DB_CONNECTION.cursor()
+ cursor.copy_expert(
+ 'COPY %s FROM STDIN WITH CSV HEADER' % table_name,
+ csv_data_io)
+ DB_CONNECTION.commit()
+
+
class SqliteRunner(object):
def __init__(self):
self.connection = sqlite3_logica.SqliteConnect()
@@ -177,13 +212,17 @@ class PostgresRunner(object):
def __init__(self):
global DB_CONNECTION
global DB_ENGINE
- if DB_CONNECTION:
- self.engine = DB_ENGINE
- self.connection = DB_CONNECTION
- else:
- (self.engine, self.connection) = PostgresJumpStart()
- DB_ENGINE = self.engine
- DB_CONNECTION = self.connection
+ if not DB_CONNECTION:
+ print("Assuming this is running on Google CoLab in a temporary")
+ print("environment.")
+ print("Would you like to install and run postgres?")
+ user_choice = input('y or N? ')
+ if user_choice != 'y':
+ print('User declined.')
+ print('Bailing out.')
+ return
+ PostgresJumpStart()
+ self.connection = DB_CONNECTION
def __call__(self, sql, engine, is_final):
return RunSQL(sql, engine, self.connection, is_final)
@@ -206,13 +245,18 @@ def Logica(line, cell, run_query):
e.ShowMessage()
return
try:
- program = universe.LogicaProgram(parsed_rules)
+ program = universe.LogicaProgram(
+ parsed_rules,
+ user_flags={'logica_default_engine': DEFAULT_ENGINE})
except functors.FunctorError as e:
e.ShowMessage()
return
except rule_translate.RuleCompileException as e:
e.ShowMessage()
return
+ except infer.TypeErrorCaughtException as e:
+ e.ShowMessage()
+ return
engine = program.annotations.Engine()
@@ -269,9 +313,13 @@ def Logica(line, cell, run_query):
else:
raise Exception('Logica only supports BigQuery, PostgreSQL and SQLite '
'for now.')
-
- result_map = concertina_lib.ExecuteLogicaProgram(
- executions, sql_runner=sql_runner, sql_engine=engine)
+ try:
+ result_map = concertina_lib.ExecuteLogicaProgram(
+ executions, sql_runner=sql_runner, sql_engine=engine,
+ display_mode=DISPLAY_MODE)
+ except infer.TypeErrorCaughtException as e:
+ e.ShowMessage()
+ return
for idx, predicate in enumerate(predicates):
t = result_map[predicate]
@@ -320,18 +368,21 @@ def PostgresJumpStart():
# Connect to the database.
from logica import colab_logica
-from sqlalchemy import create_engine
-import pandas
-engine = create_engine('postgresql+psycopg2://logica:logica@127.0.0.1', pool_recycle=3600);
-connection = engine.connect();
-colab_logica.SetDbConnection(connection)""")
+import psycopg2
+connection = psycopg2.connect(host='localhost', database='logica', user='logica', password='logica')
+connection.autocommit = True
+colab_logica.DEFAULT_ENGINE = 'psql'
+colab_logica.SetDbConnection(connection)
+""")
return
print('Installation succeeded. Connecting...')
# Connect to the database.
- from logica import colab_logica
- from sqlalchemy import create_engine
- import pandas
- engine = create_engine('postgresql+psycopg2://logica:logica@127.0.0.1', pool_recycle=3600)
- connection = engine.connect()
+ import psycopg2
+ connection = psycopg2.connect(host='localhost', database='logica', user='logica', password='logica')
+ connection.autocommit = True
+
print('Connected.')
- return engine, connection
+ global DEFAULT_ENGINE
+ global DB_CONNECTION
+ DEFAULT_ENGINE = 'psql'
+ DB_CONNECTION = connection
diff --git a/common/concertina_lib.py b/common/concertina_lib.py
index 94a96741..e4ff2b4c 100644
--- a/common/concertina_lib.py
+++ b/common/concertina_lib.py
@@ -4,10 +4,18 @@
try:
import graphviz
+except:
+ pass
+ # This is annoying to see in terminal each time.
+ # Consider adding back if lack of messaging is confusing.
+ # print('Could not import graphviz tools in Concertina.')
+
+try:
+ from IPython.display import HTML
from IPython.display import display
from IPython.display import update_display
except:
- print('Could not import CoLab tools in Concertina.')
+ print('Could not import IPython in Concertina.')
if '.' not in __package__:
from common import graph_art
@@ -33,7 +41,7 @@ def Run(self, action):
is_final=(predicate in self.final_predicates))
end = datetime.datetime.now()
if self.print_running_predicate:
- print(' (%d seconds)' % (end - start).seconds)
+ print(' (%d ms)' % int((end - start).total_seconds() * 1000))
if predicate in self.final_predicates:
self.final_result[predicate] = result
@@ -75,7 +83,7 @@ def __init__(self, config, engine, display_mode='colab'):
self.all_actions = {a["name"] for a in self.config}
self.complete_actions = set()
self.running_actions = set()
- assert display_mode in ('colab', 'terminal'), (
+ assert display_mode in ('colab', 'terminal', 'colab-text'), (
'Unrecognized display mode: %s' % display_mode)
self.display_mode = display_mode
self.display_id = self.GetDisplayId()
@@ -137,7 +145,14 @@ def AsNodesAndEdges(self):
"""Nodes and edges to display in terminal."""
def ColoredNode(node):
if node in self.running_actions:
- return '\033[1m\033[93m' + node + '\033[0m'
+ if self.display_mode == 'terminal':
+ return '\033[1m\033[93m' + node + '\033[0m'
+ elif self.display_mode == 'colab-text':
+ return (
+ '' + node + ''
+ )
+ else:
+ assert False, self.display_mode
else:
return node
nodes = []
@@ -150,11 +165,24 @@ def ColoredNode(node):
edges.append([prerequisite_node, a_node])
return nodes, edges
+ def StateAsSimpleHTML(self):
+ style = ';'.join([
+ 'border: 1px solid rgba(0, 0, 0, 0.3)',
+ 'width: fit-content;',
+ 'padding: 20px',
+ 'border-radius: 5px',
+ 'box-shadow: 1px 1px 3px rgba(0, 0, 0, 0.2)'])
+ return HTML('' % (
+ style, self.AsTextPicture(updating=False)))
+
def Display(self):
if self.display_mode == 'colab':
display(self.AsGraphViz(), display_id=self.display_id)
elif self.display_mode == 'terminal':
print(self.AsTextPicture(updating=False))
+ elif self.display_mode == 'colab-text':
+ display(self.StateAsSimpleHTML(),
+ display_id=self.display_id)
else:
assert 'Unexpected mode:', self.display_mode
@@ -163,6 +191,10 @@ def UpdateDisplay(self):
update_display(self.AsGraphViz(), display_id=self.display_id)
elif self.display_mode == 'terminal':
print(self.AsTextPicture(updating=True))
+ elif self.display_mode == 'colab-text':
+ update_display(
+ self.StateAsSimpleHTML(),
+ display_id=self.display_id)
else:
assert 'Unexpected mode:', self.display_mode
@@ -261,10 +293,14 @@ def ConcertinaConfig(table_to_export_map, dependency_edges,
print_running_predicate=(display_mode != 'terminal'))
preambles = set(e.preamble for e in logica_executions)
- assert len(preambles) == 1, 'Inconsistent preambles: %s' % preambles
- [preamble] = list(preambles)
- if preamble:
- sql_runner(preamble, sql_engine, is_final=False)
+ # Due to change of types from predicate to predicate preables are not
+ # consistent. However we expect preambles to be idempotent.
+ # So we simply run all of them.
+ # assert len(preambles) == 1, 'Inconsistent preambles: %s' % preambles
+ # [preamble] = list(preambles)
+ for preamble in preambles:
+ if preamble:
+ sql_runner(preamble, sql_engine, is_final=False)
concertina = Concertina(config, engine, display_mode=display_mode)
concertina.Run()
diff --git a/common/logica_test.py b/common/logica_test.py
index 987027ec..ee04ac2f 100644
--- a/common/logica_test.py
+++ b/common/logica_test.py
@@ -18,13 +18,18 @@
"""Utilities for YotaQL tests."""
import subprocess
+import json
if '.' not in __package__:
from common import color
from common import logica_lib
+ from type_inference.research import infer
+ from parser_py import parse
else:
from ..common import color
from ..common import logica_lib
+ from ..type_inference.research import infer
+ from ..parser_py import parse
class TestManager(object):
@@ -54,6 +59,47 @@ def RunTest(cls, name, src, predicate, golden, user_flags,
cls.GOLDEN_RUN, cls.ANNOUNCE_TESTS,
import_root)
+ @classmethod
+ def RunTypesTest(cls, name, src=None, golden=None):
+ if cls.RUN_ONLY and name not in cls.RUN_ONLY:
+ return
+ RunTypesTest(name, src, golden,
+ overwrite=cls.GOLDEN_RUN)
+
+
+def RunTypesTest(name, src=None, golden=None,
+ overwrite=False):
+ src = src or (name + '.l')
+ golden = golden or (name + '.txt')
+
+ test_result = '{warning}RUNNING{end}'
+ print(color.Format('% 50s %s' % (name, test_result)))
+
+ program_text = open(src).read()
+ try:
+ parsed_rules = parse.ParseFile(program_text)['rule']
+ except parse.ParsingException as parsing_exception:
+ parsing_exception.ShowMessage()
+ sys.exit(1)
+
+ typing_engine = infer.TypesInferenceEngine(parsed_rules)
+ typing_engine.InferTypes()
+ result = json.dumps(parsed_rules, sort_keys=True, indent=' ')
+
+ if overwrite:
+ with open(golden, 'w') as w:
+ w.write(result)
+ golden_result = open(golden).read()
+
+ if result == golden_result:
+ test_result = '{ok}PASSED{end}'
+ else:
+ p = subprocess.Popen(['diff', '-', golden], stdin=subprocess.PIPE)
+ p.communicate(result.encode())
+ test_result = '{error}FAILED{end}'
+
+ print('\033[F\033[K' + color.Format('% 50s %s' % (name, test_result)))
+
def RunTest(name, src, predicate, golden,
user_flags=None,
diff --git a/common/psql_logica.py b/common/psql_logica.py
new file mode 100644
index 00000000..443aeda6
--- /dev/null
+++ b/common/psql_logica.py
@@ -0,0 +1,94 @@
+#!/usr/bin/python
+#
+# Copyright 2023 Google LLC
+#
+# Licensed 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 getpass
+import json
+import os
+import re
+from decimal import Decimal
+
+if '.' not in __package__:
+ from type_inference.research import infer
+else:
+ from ..type_inference.research import infer
+
+
+def PostgresExecute(sql, connection):
+ import psycopg2
+ import psycopg2.extras
+ cursor = connection.cursor()
+ try:
+ cursor.execute(sql)
+ # Make connection aware of the used types.
+ types = re.findall(r'-- Logica type: (\w*)', sql)
+ for t in types:
+ if t != 'logicarecord893574736': # Empty record.
+ psycopg2.extras.register_composite(t, cursor, globally=True)
+ except psycopg2.errors.UndefinedTable as e:
+ raise infer.TypeErrorCaughtException(
+ infer.ContextualizedError.BuildNiceMessage(
+ 'Running SQL.', 'Undefined table used: ' + str(e)))
+ except psycopg2.Error as e:
+ connection.rollback()
+ raise e
+ return cursor
+
+
+def DigestPsqlType(x):
+ if isinstance(x, tuple):
+ return PsqlTypeAsDictionary(x)
+ if isinstance(x, list) and len(x) > 0:
+ return PsqlTypeAsList(x)
+ if isinstance(x, Decimal):
+ if x.as_integer_ratio()[1] == 1:
+ return int(x)
+ else:
+ return float(x)
+ return x
+
+
+def PsqlTypeAsDictionary(record):
+ result = {}
+ for f in record._asdict():
+ a = getattr(record, f)
+ result[f] = DigestPsqlType(a)
+ return result
+
+
+def PsqlTypeAsList(a):
+ return list(map(DigestPsqlType, a))
+
+
+def ConnectToPostgres(mode):
+ import psycopg2
+ if mode == 'interactive':
+ print('Please enter PostgreSQL URL, or config in JSON format with fields host, database, user and password.')
+ connection_str = getpass.getpass()
+ elif mode == 'environment':
+ connection_str = os.environ.get('LOGICA_PSQL_CONNECTION')
+ assert connection_str, (
+ 'Please provide PSQL connection parameters '
+ 'in LOGICA_PSQL_CONNECTION.')
+ else:
+ assert False, 'Unknown mode:' + mode
+ if connection_str.startswith('postgres'):
+ connection = psycopg2.connect(connection_str)
+ else:
+ connection_json = json.loads(connection_str)
+ connection = psycopg2.connect(**connection_json)
+
+ connection.autocommit = True
+ return connection
\ No newline at end of file
diff --git a/compiler/dialect_libraries/bq_library.py b/compiler/dialect_libraries/bq_library.py
index d6cc3546..6c794716 100644
--- a/compiler/dialect_libraries/bq_library.py
+++ b/compiler/dialect_libraries/bq_library.py
@@ -17,6 +17,7 @@
library = """
->(left:, right:) = {arg: left, value: right};
`=`(left:, right:) = right :- left == right;
+`~`(left:, right:); # No action. Compiler unifies types.
# All ORDER BY arguments are wrapped, to avoid confusion with
# column index.
diff --git a/compiler/dialect_libraries/presto_library.py b/compiler/dialect_libraries/presto_library.py
index 90da11d4..c7015bc4 100644
--- a/compiler/dialect_libraries/presto_library.py
+++ b/compiler/dialect_libraries/presto_library.py
@@ -17,6 +17,7 @@
library = """
->(left:, right:) = {arg: left, value: right};
`=`(left:, right:) = right :- left == right;
+`~`(left:, right:); # No action. Compiler unifies types.
ArgMin(a) = SqlExpr("(ARRAY_AGG({arg} order by {value}))[1]",
{arg: a.arg, value: a.value});
diff --git a/compiler/dialect_libraries/psql_library.py b/compiler/dialect_libraries/psql_library.py
index 2bd1e31b..db52216b 100644
--- a/compiler/dialect_libraries/psql_library.py
+++ b/compiler/dialect_libraries/psql_library.py
@@ -17,13 +17,14 @@
library = """
->(left:, right:) = {arg: left, value: right};
`=`(left:, right:) = right :- left == right;
+`~`(left:, right:); # No action. Compiler unifies types.
-ArgMin(a) = SqlExpr("(ARRAY_AGG({arg} order by {value}))[1]",
- {arg: a.arg, value: a.value});
+ArgMin(a) = (SqlExpr("(ARRAY_AGG({arg} order by {value}))[1]",
+ {arg: {argpod: a.arg}, value: a.value})).argpod;
-ArgMax(a) = SqlExpr(
+ArgMax(a) = (SqlExpr(
"(ARRAY_AGG({arg} order by {value} desc))[1]",
- {arg: a.arg, value: a.value});
+ {arg: {argpod: a.arg}, value: a.value})).argpod;
ArgMaxK(a, l) = SqlExpr(
"(ARRAY_AGG({arg} order by {value} desc))[1:{lim}]",
@@ -37,4 +38,11 @@
"ARRAY_AGG({value} order by {arg})",
{arg: a.arg, value: a.value});
+RecordAsJson(r) = SqlExpr(
+ "ROW_TO_JSON({r})", {r:});
+
+Fingerprint(s) = SqlExpr("('x' || substr(md5({s}), 1, 16))::bit(64)::bigint", {s:});
+
+Num(a) = a;
+Str(a) = a;
"""
diff --git a/compiler/dialect_libraries/sqlite_library.py b/compiler/dialect_libraries/sqlite_library.py
index 284a94d7..1a4f4eb0 100644
--- a/compiler/dialect_libraries/sqlite_library.py
+++ b/compiler/dialect_libraries/sqlite_library.py
@@ -17,6 +17,7 @@
library = """
->(left:, right:) = {arg: left, value: right};
`=`(left:, right:) = right :- left == right;
+`~`(left:, right:); # No action. Compiler unifies types.
Arrow(left, right) = arrow :-
left == arrow.arg,
diff --git a/compiler/dialect_libraries/trino_library.py b/compiler/dialect_libraries/trino_library.py
index 90da11d4..c7015bc4 100644
--- a/compiler/dialect_libraries/trino_library.py
+++ b/compiler/dialect_libraries/trino_library.py
@@ -17,6 +17,7 @@
library = """
->(left:, right:) = {arg: left, value: right};
`=`(left:, right:) = right :- left == right;
+`~`(left:, right:); # No action. Compiler unifies types.
ArgMin(a) = SqlExpr("(ARRAY_AGG({arg} order by {value}))[1]",
{arg: a.arg, value: a.value});
diff --git a/compiler/dialects.py b/compiler/dialects.py
index c1e0950e..a281f821 100755
--- a/compiler/dialects.py
+++ b/compiler/dialects.py
@@ -40,6 +40,13 @@ def Get(engine):
class Dialect(object):
pass
+ # Default methods:
+ def MaybeCascadingDeletionWord(self):
+ return '' # No CASCADE is needed by default.
+
+ def PredicateLiteral(self, predicate_name):
+ return "'predicate_name:%s'" % predicate_name
+
class BigQueryDialect(Dialect):
"""BigQuery SQL dialect."""
@@ -55,7 +62,7 @@ def InfixOperators(self):
'++': 'CONCAT(%s, %s)',
}
- def Subscript(self, record, subscript):
+ def Subscript(self, record, subscript, record_is_table):
return '%s.%s' % (record, subscript)
def LibraryProgram(self):
@@ -73,6 +80,10 @@ def GroupBySpecBy(self):
def DecorateCombineRule(self, rule, var):
return rule
+ def PredicateLiteral(self, predicate_name):
+ return 'STRUCT("%s" AS predicate_name)' % predicate_name
+
+
class SqLiteDialect(Dialect):
"""SqLite SQL dialect."""
@@ -102,74 +113,7 @@ def BuiltInFunctions(self):
}
def DecorateCombineRule(self, rule, var):
- """Resolving ambiguity of aggregation scope."""
- # Entangling result of aggregation with a variable that comes from a list
- # unnested inside a combine expression, to make it clear that aggregation
- # must be done in the combine.
- rule = copy.deepcopy(rule)
-
- rule['head']['record']['field_value'][0]['value'][
- 'aggregation']['expression']['call'][
- 'record']['field_value'][0]['value'] = (
- {
- 'expression': {
- 'call': {
- 'predicate_name': 'MagicalEntangle',
- 'record': {
- 'field_value': [
- {
- 'field': 0,
- 'value': rule['head']['record']['field_value'][0]['value'][
- 'aggregation']['expression']['call'][
- 'record']['field_value'][0]['value']
- },
- {
- 'field': 1,
- 'value': {
- 'expression': {
- 'variable': {
- 'var_name': var
- }
- }
- }
- }
- ]
- }
- }
- }
- }
- )
-
- if 'body' not in rule:
- rule['body'] = {'conjunction': {'conjunct': []}}
- rule['body']['conjunction']['conjunct'].append(
- {
- "inclusion": {
- "list": {
- "literal": {
- "the_list": {
- "element": [
- {
- "literal": {
- "the_number": {
- "number": "0"
- }
- }
- }
- ]
- }
- }
- },
- "element": {
- "variable": {
- "var_name": var
- }
- }
- }
- }
- )
- return rule
-
+ return DecorateCombineRule(rule, var)
def InfixOperators(self):
return {
@@ -178,9 +122,12 @@ def InfixOperators(self):
'in': 'IN_LIST(%s, %s)'
}
- def Subscript(self, record, subscript):
- return 'JSON_EXTRACT(%s, "$.%s")' % (record, subscript)
-
+ def Subscript(self, record, subscript, record_is_table):
+ if record_is_table:
+ return '%s.%s' % (record, subscript)
+ else:
+ return 'JSON_EXTRACT(%s, "$.%s")' % (record, subscript)
+
def LibraryProgram(self):
return sqlite_library.library
@@ -203,9 +150,13 @@ def BuiltInFunctions(self):
return {
'Range': '(SELECT ARRAY_AGG(x) FROM GENERATE_SERIES(0, {0} - 1) as x)',
'ToString': 'CAST(%s AS TEXT)',
+ 'ToInt64': 'CAST(%s AS BIGINT)',
'Element': '({0})[{1} + 1]',
- 'Size': 'ARRAY_LENGTH(%s, 1)',
- 'Count': 'COUNT(DISTINCT {0})'
+ 'Size': 'COALESCE(ARRAY_LENGTH({0}, 1), 0)',
+ 'Count': 'COUNT(DISTINCT {0})',
+ 'MagicalEntangle': '(CASE WHEN {1} = 0 THEN {0} ELSE NULL END)',
+ 'ArrayConcat': '{0} || {1}',
+ 'Split': 'STRING_TO_ARRAY({0}, {1})'
}
def InfixOperators(self):
@@ -213,7 +164,7 @@ def InfixOperators(self):
'++': 'CONCAT(%s, %s)',
}
- def Subscript(self, record, subscript):
+ def Subscript(self, record, subscript, record_is_table):
return '(%s).%s' % (record, subscript)
def LibraryProgram(self):
@@ -226,10 +177,13 @@ def ArrayPhrase(self):
return 'ARRAY[%s]'
def GroupBySpecBy(self):
- return 'name'
+ return 'expr'
def DecorateCombineRule(self, rule, var):
- return rule
+ return DecorateCombineRule(rule, var)
+
+ def MaybeCascadingDeletionWord(self):
+ return ' CASCADE' # Need to cascade in PSQL.
class Trino(Dialect):
@@ -253,7 +207,7 @@ def InfixOperators(self):
'++': 'CONCAT(%s, %s)',
}
- def Subscript(self, record, subscript):
+ def Subscript(self, record, subscript, record_is_table):
return '%s.%s' % (record, subscript)
def LibraryProgram(self):
@@ -291,7 +245,7 @@ def InfixOperators(self):
'++': 'CONCAT(%s, %s)',
}
- def Subscript(self, record, subscript):
+ def Subscript(self, record, subscript, record_is_table):
return '%s.%s' % (record, subscript)
def LibraryProgram(self):
@@ -309,6 +263,74 @@ def GroupBySpecBy(self):
def DecorateCombineRule(self, rule, var):
return rule
+def DecorateCombineRule(rule, var):
+ """Resolving ambiguity of aggregation scope."""
+ # Entangling result of aggregation with a variable that comes from a list
+ # unnested inside a combine expression, to make it clear that aggregation
+ # must be done in the combine.
+ rule = copy.deepcopy(rule)
+
+ rule['head']['record']['field_value'][0]['value'][
+ 'aggregation']['expression']['call'][
+ 'record']['field_value'][0]['value'] = (
+ {
+ 'expression': {
+ 'call': {
+ 'predicate_name': 'MagicalEntangle',
+ 'record': {
+ 'field_value': [
+ {
+ 'field': 0,
+ 'value': rule['head']['record']['field_value'][0]['value'][
+ 'aggregation']['expression']['call'][
+ 'record']['field_value'][0]['value']
+ },
+ {
+ 'field': 1,
+ 'value': {
+ 'expression': {
+ 'variable': {
+ 'var_name': var
+ }
+ }
+ }
+ }
+ ]
+ }
+ }
+ }
+ }
+ )
+
+ if 'body' not in rule:
+ rule['body'] = {'conjunction': {'conjunct': []}}
+ rule['body']['conjunction']['conjunct'].append(
+ {
+ "inclusion": {
+ "list": {
+ "literal": {
+ "the_list": {
+ "element": [
+ {
+ "literal": {
+ "the_number": {
+ "number": "0"
+ }
+ }
+ }
+ ]
+ }
+ }
+ },
+ "element": {
+ "variable": {
+ "var_name": var
+ }
+ }
+ }
+ }
+ )
+ return rule
class Databricks(Dialect):
"""Databricks dialect"""
diff --git a/compiler/expr_translate.py b/compiler/expr_translate.py
index 4b7ef5c9..c9bb299a 100755
--- a/compiler/expr_translate.py
+++ b/compiler/expr_translate.py
@@ -39,6 +39,7 @@ class QL(object):
'ToUInt64': 'CAST(%s AS UINT64)',
'ToString': 'CAST(%s AS STRING)',
# Aggregation.
+ '1': 'MIN(%s)', # Should be ANY_VALUE, but using MIN, so it works everywhere.
'Aggr': '%s', # Placeholder to use formulas for aggregation.
'Agg+': 'SUM(%s)',
'Agg++': 'ARRAY_CONCAT_AGG(%s)',
@@ -96,8 +97,8 @@ class QL(object):
'&&': '%s AND %s',
'%': 'MOD(%s, %s)'
}
- BULK_FUNCTIONS = {}
- BULK_FUNCTIONS_ARITY_RANGE = {}
+ BULK_FUNCTIONS = None
+ BULK_FUNCTIONS_ARITY_RANGE = None
# When adding any analytic functions please check that ConvertAnalytic
# function handles them correctly.
@@ -188,18 +189,22 @@ def CamelCase(s):
reader = processed_functions.GetCsv()
header = next(reader)
+ bulk_functions = {}
+ bulk_functions_arity_range = {}
for row in reader:
row = dict(list(zip(header, row)))
if row['function'][0] == '$':
# TODO: Process operators.
continue
function_name = CamelCase(row['function'])
- cls.BULK_FUNCTIONS[function_name] = (
+ bulk_functions[function_name] = (
'%s(%s)' % (row['sql_function'], '%s'))
- cls.BULK_FUNCTIONS_ARITY_RANGE[function_name] = (
+ bulk_functions_arity_range[function_name] = (
int(row['min_args']),
float('inf')
if row['has_repeated_args'] == '1' else int(row['max_args']))
+ cls.BULK_FUNCTIONS = bulk_functions
+ cls.BULK_FUNCTIONS_ARITY_RANGE = bulk_functions_arity_range
def BuiltInFunctionArityRange(self, f):
"""Returns arity of the built-in function."""
@@ -235,10 +240,10 @@ def Function(self, f, args):
def Infix(self, op, args):
return op % (args['left'], args['right'])
- def Subscript(self, record, subscript):
+ def Subscript(self, record, subscript, record_is_table):
if isinstance(subscript, int):
subscript = 'col%d' % subscript
- return self.dialect.Subscript(record, subscript)
+ return self.dialect.Subscript(record, subscript, record_is_table)
def IntLiteral(self, literal):
return str(literal['number'])
@@ -254,8 +259,13 @@ def ListLiteralInternals(self, literal):
return ', '.join([self.ConvertToSql(e)
for e in literal['element']])
- def ListLiteral(self, literal):
- return self.dialect.ArrayPhrase() % self.ListLiteralInternals(literal)
+ def ListLiteral(self, literal, element_type_name):
+ suffix = ('::' + element_type_name + '[]'
+ if self.dialect.Name() == 'PostgreSQL'
+ else '')
+ return (
+ self.dialect.ArrayPhrase() %
+ self.ListLiteralInternals(literal)) + suffix
def BoolLiteral(self, literal):
return literal['the_bool']
@@ -267,9 +277,35 @@ def NullLiteral(self, literal):
def PredicateLiteral(self, literal):
if self.convert_to_json:
return '{"predicate_name": "%s"}' % (literal['predicate_name'])
- return 'STRUCT("%s" AS predicate_name)' % literal['predicate_name']
-
- def Variable(self, variable):
+ return self.dialect.PredicateLiteral(literal['predicate_name'])
+
+ def VariableMaybeTableSQLite(self, variable, expression_type):
+ def MakeSubscript(subscripted_variable, subscripted_field):
+ return {'expression':
+ {'subscript': {'record': {'variable':
+ {'var_name': subscripted_variable,
+ 'dont_expand': True}},
+ 'subscript': {
+ 'literal': {'the_symbol': {
+ 'symbol': subscripted_field}}}}}}
+ expr = self.vocabulary[variable['var_name']]
+ if '.' not in expr and 'dont_expand' not in variable:
+
+ if not isinstance(expression_type, dict):
+ raise self.exception_maker(
+ 'Could not create record ' + color.Warn(expr) +
+ '. Type inference is ' +
+ 'required to convert table rows to records in SQLite.')
+ return self.ConvertToSql({'record': {'field_value': [
+ {'field': k,
+ 'value': MakeSubscript(variable['var_name'], k)}
+ for k in sorted(expression_type)
+ ]}})
+ return expr
+
+ def Variable(self, variable, expression_type):
+ if self.dialect.Name() == 'SqLite':
+ return self.VariableMaybeTableSQLite(variable, expression_type)
if variable['var_name'] in self.vocabulary:
return self.vocabulary[variable['var_name']]
else:
@@ -294,7 +330,7 @@ def RecordAsJson(self, record):
value=self.ConvertToSql(f_v['value']['expression'])))
return '{%s}' % ', '.join(json_field_values)
- def Record(self, record):
+ def Record(self, record, record_type=None):
if self.convert_to_json:
return self.RecordAsJson(record)
# TODO: Move this to dialects.py.
@@ -310,6 +346,12 @@ def Record(self, record):
for f_v in record['field_value'])
if self.dialect.Name() == 'Trino':
return '(SELECT %s)' % arguments_str
+ if self.dialect.Name() == 'PostgreSQL':
+ assert record_type, json.dumps(record, indent=' ')
+ args = ', '.join(
+ self.ConvertToSql(f_v['value']['expression'])
+ for f_v in sorted(record['field_value'], key=lambda x: x['field']))
+ return 'ROW(%s)::%s' % (args, record_type)
return 'STRUCT(%s)' % arguments_str
def GenericSqlExpression(self, record):
@@ -418,12 +460,29 @@ def GetValueOfField(field_values, field):
'implication': {'if_then': new_if_thens, 'otherwise': new_otherwise}}
return self.ConvertToSql(new_expr)
+ def ConvertToSqlForGroupBy(self, expression):
+ if 'literal' in expression and 'the_string' in expression['literal']:
+ # To calm down PSQL:
+ return f"({self.ConvertToSql(expression)} || '')"
+ return self.ConvertToSql(expression)
+
+ def ExpressionIsTable(self, expression):
+ return (
+ 'variable' in expression and
+ 'dont_expand' in expression['variable'] and
+ self.VariableIsTable(expression['variable']['var_name']))
+
+ def VariableIsTable(self, variable_name):
+ return '.' not in self.vocabulary[variable_name]
+
def ConvertToSql(self, expression):
"""Converting Logica expression into SQL."""
# print('EXPR:', expression)
# Variables.
if 'variable' in expression:
- return self.Variable(expression['variable'])
+ the_type = expression.get('type', {}).get('the_type', 'Any')
+ return self.Variable(expression['variable'],
+ the_type)
# Literals.
if 'literal' in expression:
@@ -433,7 +492,14 @@ def ConvertToSql(self, expression):
if 'the_string' in literal:
return self.StrLiteral(literal['the_string'])
if 'the_list' in literal:
- return self.ListLiteral(literal['the_list'])
+ the_list = literal['the_list']
+ element_type = expression.get('type', {}).get('element_type_name', None)
+ if self.dialect.Name() == 'PostgreSQL' and element_type is None:
+ raise self.exception_maker(color.Format(
+ 'Array needs type in PostgreSQL: '
+ '{warning}{the_list}{end}.', dict(
+ the_list=expression['expression_heritage'])))
+ return self.ListLiteral(the_list, element_type)
if 'the_bool' in literal:
return self.BoolLiteral(literal['the_bool'])
if 'the_null' in literal:
@@ -540,11 +606,23 @@ def ConvertToSql(self, expression):
return simplified_sub
# Couldn't optimize, just return the '.' expression.
record = self.ConvertToSql(sub['record'])
- return self.Subscript(record, subscript)
+ return self.Subscript(record, subscript, self.ExpressionIsTable(sub['record']))
if 'record' in expression:
record = expression['record']
- return self.Record(record)
+ record_type = expression.get('type', {}).get('type_name', None)
+ if self.dialect.Name() == 'PostgreSQL' and record_type is None:
+ rendered_type = expression.get('type', {}).get('rendered_type', None)
+ raise self.exception_maker(color.Format(
+ 'Record needs type in PostgreSQL: '
+ '{warning}{record}{end} was inferred only '
+ 'an incomplete type {warning}{type}{end}.', dict(
+ record=expression['expression_heritage'],
+ type=rendered_type)))
+
+ return self.Record(
+ record,
+ record_type=record_type)
if 'combine' in expression:
return '(%s)' % (
diff --git a/compiler/functors.py b/compiler/functors.py
index 7e5ded87..e04a4c16 100755
--- a/compiler/functors.py
+++ b/compiler/functors.py
@@ -42,6 +42,16 @@
from ..compiler.dialect_libraries import recursion_library
from ..parser_py import parse
+import datetime
+
+class Timer:
+ def __init__(self, name):
+ self.name = name
+ self.time = datetime.datetime.now()
+
+ def Stop(self):
+ print('%s / micros elapsed:' % self.name, (datetime.datetime.now() - self.time).microseconds)
+
class FunctorError(Exception):
"""Exception thrown when Make is bad."""
@@ -172,9 +182,18 @@ def BuildArgs(self, functor):
while queue:
e = queue.popleft()
result.add(e)
- for a in self.ArgsOf(e):
+ args_of_e = self.ArgsOf(e)
+ if isinstance(args_of_e, set):
+ arg_type = 'final'
+ else:
+ arg_type = 'preliminary'
+
+ for a in args_of_e:
if a not in result:
- queue.append(a)
+ if arg_type == 'preliminary':
+ queue.append(a)
+ else:
+ result.add(a)
del self.args_of[functor]
diff --git a/compiler/rule_translate.py b/compiler/rule_translate.py
index d2072917..cdd95541 100755
--- a/compiler/rule_translate.py
+++ b/compiler/rule_translate.py
@@ -206,6 +206,13 @@ def __init__(self, names_allocator=None, external_vocabulary=None,
self.full_rule_text = None
self.distinct_denoted = None
+ def SelectAsRecord(self):
+ return {'record': {
+ 'field_value': [{
+ 'field': k,
+ 'value': {'expression': v}
+ } for k, v in sorted(self.select.items())]}}
+
def OwnVarsVocabulary(self):
"""Returns a map: logica variable -> SQL expression with the value."""
def TableAndFieldToSql(table, field):
@@ -284,7 +291,8 @@ def ReplaceVariableEverywhere(self, u_left, u_right):
ReplaceVariable(u_left, u_right, self.vars_unification)
ReplaceVariable(u_left, u_right, self.constraints)
- def ElliminateInternalVariables(self, assert_full_ellimination=False):
+ # TODO: Parameter unfold_recods just patches some bug. Careful review is needed.
+ def ElliminateInternalVariables(self, assert_full_ellimination=False, unfold_records=True):
"""Elliminates internal variables via substitution."""
variables = self.InternalVariables()
while True:
@@ -309,7 +317,9 @@ def ElliminateInternalVariables(self, assert_full_ellimination=False):
self.ReplaceVariableEverywhere(u_left, u_right)
done = False
# Assignments to variables in record fields.
- if True: # Confirm that unwraping works and make this unconditional.
+ if unfold_records: # Confirm that unwraping works and make this unconditional.
+ # Unwrapping goes wild sometimes. Letting it go right to left only.
+ # for k, r in [['left', 'right']]:
for k, r in [['left', 'right'], ['right', 'left']]:
if u[k] == u[r]:
continue
@@ -476,7 +486,11 @@ def AsSql(self, subquery_encoder=None, flag_values=None):
for k, v in self.select.items():
if k == '*':
- fields.append('%s.*' % ql.ConvertToSql(v))
+ if 'variable' in v:
+ v['variable']['dont_expand'] = True # For SQLite.
+ fields.append(
+ subquery_encoder.execution.dialect.Subscript(
+ ql.ConvertToSql(v), '*', True))
else:
fields.append('%s AS %s' % (ql.ConvertToSql(v), LogicaFieldToSqlField(k)))
r += ',\n'.join(' ' + f for f in fields)
@@ -504,11 +518,14 @@ def AsSql(self, subquery_encoder=None, flag_values=None):
tables.append(sql)
self.SortUnnestings()
for element, the_list in self.unnestings:
+ if 'variable' in element:
+ # To prevent SQLite record unfolding.
+ element['variable']['dont_expand'] = True
tables.append(
subquery_encoder.execution.dialect.UnnestPhrase().format(
ql.ConvertToSql(the_list), ql.ConvertToSql(element)))
if not tables:
- tables.append('(SELECT "singleton" as s) as unused_singleton')
+ tables.append("(SELECT 'singleton' as s) as unused_singleton")
from_str = ', '.join(tables)
# Indent the from_str.
from_str = '\n'.join(' ' + l for l in from_str.split('\n'))
@@ -531,7 +548,8 @@ def AsSql(self, subquery_encoder=None, flag_values=None):
for v in ordered_distinct_vars)
elif subquery_encoder.execution.dialect.GroupBySpecBy() == 'expr':
r += ', '.join(
- ql.ConvertToSql(self.select[k]) for k in ordered_distinct_vars
+ ql.ConvertToSqlForGroupBy(self.select[k])
+ for k in ordered_distinct_vars
)
else:
assert False, 'Broken dialect %s, group by spec: %s' % (
@@ -609,7 +627,8 @@ def ExtractInclusionStructure(inclusion, s):
'value': {
'expression': {
'variable': {
- 'var_name': var_name
+ 'var_name': var_name,
+ 'dont_expand': True
}
}
}
diff --git a/compiler/universe.py b/compiler/universe.py
index fb4673ff..f29452cb 100755
--- a/compiler/universe.py
+++ b/compiler/universe.py
@@ -34,6 +34,7 @@
from compiler import functors
from compiler import rule_translate
from parser_py import parse
+ from type_inference.research import infer
else:
from ..common import color
from ..compiler import dialects
@@ -41,6 +42,7 @@
from ..compiler import functors
from ..compiler import rule_translate
from ..parser_py import parse
+ from ..type_inference.research import infer
PredicateInfo = collections.namedtuple('PredicateInfo',
['embeddable'])
@@ -135,6 +137,11 @@ class Annotations(object):
def __init__(self, rules, user_flags):
# Extracting DefineFlags first, so that we can use flags in @Ground
# annotations.
+ if 'logica_default_engine' in user_flags:
+ self.default_engine = user_flags['logica_default_engine']
+ else:
+ self.default_engine = 'bigquery'
+
self.annotations = self.ExtractAnnotations(
rules, restrict_to=['@DefineFlag', '@ResetFlagValue'])
self.user_flags = user_flags
@@ -154,7 +161,7 @@ def Preamble(self):
preamble += (
'-- Initializing PostgreSQL environment.\n'
'set client_min_messages to warning;\n'
- 'create schema if not exists logica_test;\n\n')
+ 'create schema if not exists logica_home;\n\n')
return preamble
def BuildFlagValues(self):
@@ -165,12 +172,14 @@ def BuildFlagValues(self):
programmatic_flag_values = {}
for flag, a in self.annotations['@ResetFlagValue'].items():
programmatic_flag_values[flag] = a.get('1', '${%s}' % flag)
+ system_flags = set(['logica_default_engine'])
+ allowed_flags_set = set(default_values) | system_flags
- if not set(self.user_flags) <= set(default_values):
+ if not set(self.user_flags) <= allowed_flags_set:
raise rule_translate.RuleCompileException(
'Undefined flags used: %s' % list(
- set(self.user_flags) - set(default_values)),
- str(set(self.user_flags) - set(default_values)))
+ set(self.user_flags) - allowed_flags_set),
+ str(set(self.user_flags) - allowed_flags_set))
flag_values = default_values
flag_values.update(**programmatic_flag_values)
flag_values.update(**self.user_flags)
@@ -192,7 +201,6 @@ def AttachedDatabases(self):
result = {}
for k, v in self.annotations['@AttachDatabase'].items():
if '1' not in v:
- print('>>', v)
AnnotationError('@AttachDatabase must have a single argument.',
v)
result[k] = v['1']
@@ -245,14 +253,34 @@ def OrderBy(self, predicate_name):
return FieldValuesAsList(self.annotations['@OrderBy'][predicate_name])
def Dataset(self):
- return self.ExtractSingleton('@Dataset', 'logica_test')
+ default_dataset = 'logica_test'
+ # This change is intended for all engines in the future.
+ if self.Engine() == 'psql':
+ default_dataset = 'logica_home'
+ return self.ExtractSingleton('@Dataset', default_dataset)
def Engine(self):
- engine = self.ExtractSingleton('@Engine', 'bigquery')
+ engine = self.ExtractSingleton('@Engine', self.default_engine)
if engine not in dialects.DIALECTS:
AnnotationError('Unrecognized engine: %s' % engine,
self.annotations['@Engine'][engine])
return engine
+
+ def ShouldTypecheck(self):
+ engine = self.Engine()
+
+ if '@Engine' not in self.annotations:
+ return engine == 'psql'
+ if len(self.annotations['@Engine'].values()) == 0:
+ return engine == 'psql'
+
+ engine_annotation = list(self.annotations['@Engine'].values())[0]
+ if 'type_checking' not in engine_annotation:
+ if engine == 'psql':
+ return True
+ else:
+ return False
+ return engine_annotation['type_checking']
def ExtractSingleton(self, annotation_name, default_value):
if not self.annotations[annotation_name]:
@@ -394,6 +422,17 @@ def __contains__(self, key):
ql.convert_to_json = True
annotation = rule['head']['predicate_name']
+ # Checking for aggregations.
+ aggregated_fields = [
+ fv['field']
+ for fv in rule['head']['record']['field_value']
+ if 'aggregation' in fv['value']]
+ if aggregated_fields:
+ raise rule_translate.RuleCompileException(
+ 'Annotation may not use aggregation, but field '
+ '%s is aggregated.' % (
+ color.Warn(aggregated_fields[0])),
+ rule_text)
field_values_json_str = ql.ConvertToSql(
{'record': rule['head']['record']})
try:
@@ -490,14 +529,20 @@ def __init__(self, rules, table_aliases=None, user_flags=None):
# We need to recompute annotations, because 'Make' created more rules and
# annotations.
self.annotations = Annotations(extended_rules, self.user_flags)
+
+ # Infering types if requested.
+ self.typing_preamble = ''
+ self.required_type_definitions = {}
+ self.predicate_signatures = {}
+ self.typing_engine = None
+ if self.annotations.ShouldTypecheck():
+ self.typing_preamble = self.RunTypechecker()
+
# Build udfs, populating custom_udfs and custom_udf_definitions.
self.BuildUdfs()
# Function compilation may have added irrelevant defines:
self.execution = None
- if False:
- self.RunTypechecker()
-
def UnfoldRecursion(self, rules):
annotations = Annotations(rules, {})
f = functors.Functors(rules)
@@ -530,7 +575,15 @@ def RunTypechecker(self):
Raises:
TypeInferenceError if there are any type errors.
"""
- inference.CheckTypes(self.rules)
+ rules = [r for _, r in self.rules]
+ typing_engine = infer.TypesInferenceEngine(rules)
+ typing_engine.InferTypes()
+ self.typing_engine = typing_engine
+ type_error_checker = infer.TypeErrorChecker(rules)
+ type_error_checker.CheckForError(mode='raise')
+ self.predicate_signatures = typing_engine.predicate_signature
+ self.required_type_definitions.update(typing_engine.collector.definitions)
+ return typing_engine.typing_preamble
def RunMakes(self, rules):
"""Runs @Make instructions."""
@@ -751,6 +804,10 @@ def InitializeExecution(self, main_predicate):
[])
self.execution.dependencies_of = self.functors.args_of
self.execution.dialect = dialects.Get(self.annotations.Engine())
+
+ def UpdateExecutionWithTyping(self):
+ if self.execution.dialect.Name() == 'PostgreSQL':
+ self.execution.preamble += '\n' + self.typing_preamble
def FormattedPredicateSql(self, name, allocator=None):
"""Printing top-level formatted SQL statement with defines and exports."""
@@ -767,6 +824,8 @@ def FormattedPredicateSql(self, name, allocator=None):
else:
sql = self.PredicateSql(name, allocator)
+ self.UpdateExecutionWithTyping()
+
assert self.execution.workflow_predicates_stack == [name], (
'Logica internal error: unexpected workflow stack: %s' %
self.execution.workflow_predicates_stack)
@@ -850,7 +909,7 @@ def RunInjections(self, s, allocator):
[r] = rules
rs = rule_translate.ExtractRuleStructure(
r, allocator, None)
- rs.ElliminateInternalVariables(assert_full_ellimination=False)
+ rs.ElliminateInternalVariables(assert_full_ellimination=False, unfold_records=False)
new_tables.update(rs.tables)
InjectStructure(s, rs)
@@ -877,6 +936,11 @@ def RunInjections(self, s, allocator):
}
}
})
+ elif table_var == '*':
+ s.vars_unification.append({
+ 'left': {'variable': {'var_name': clause_var}},
+ 'right': rs.SelectAsRecord()
+ })
else:
extra_hint = '' if table_var != '*' else (
' Are you using .. for injectible predicate? '
@@ -915,11 +979,19 @@ def SingleRuleSql(self, rule,
s = rule_translate.ExtractRuleStructure(
r, allocator, external_vocabulary)
- s.ElliminateInternalVariables(assert_full_ellimination=False)
+ # TODO(2023 July): Was this always redundant?
+ # s.ElliminateInternalVariables(assert_full_ellimination=False)
self.RunInjections(s, allocator)
s.ElliminateInternalVariables(assert_full_ellimination=True)
s.UnificationsToConstraints()
+ type_inference = infer.TypeInferenceForStructure(s, self.predicate_signatures)
+ type_inference.PerformInference()
+ # New types may arrive here when we have an injetible predicate with variables
+ # which specific record type depends on the inputs.
+ self.required_type_definitions.update(type_inference.collector.definitions)
+ self.typing_preamble = infer.BuildPreamble(self.required_type_definitions)
+
try:
sql = s.AsSql(self.MakeSubqueryTranslator(allocator), self.flag_values)
except RuntimeError as runtime_error:
@@ -987,8 +1059,9 @@ def TranslateTableAttachedToFile(self, table, ground, external_vocabulary):
dependency_sql = self.program.UseFlagsAsParameters(dependency_sql)
self.execution.workflow_predicates_stack.pop()
maybe_drop_table = (
- 'DROP TABLE IF EXISTS %s;\n' % ground.table_name
- if ground.overwrite else '')
+ 'DROP TABLE IF EXISTS %s%s;\n' % ((
+ ground.table_name if ground.overwrite else '',
+ self.execution.dialect.MaybeCascadingDeletionWord())))
export_statement = (
maybe_drop_table +
'CREATE TABLE {name} AS {dependency_sql}'.format(
diff --git a/docs/index.html b/docs/index.html
index 37ecd0b0..0242a23c 100644
--- a/docs/index.html
+++ b/docs/index.html
@@ -329,7 +329,8 @@ Why Logica?
Logica is for engineers, data scientists and other specialists who
need to perform complex data processing and analysis.
Queries and pipelines written in Logica can run on
-BigQuery and SQLite engines.
+BigQuery, SQLite
+and PostgreSQL engines.
Information stored in these systems is thus available in Logica.
@@ -347,11 +348,11 @@ Why Logica?
- you already use logic programming and need more computational power, or
- - you use SQL, but feel unsatisfied about its readability, or
+ - you already have data in BigQuery, PostgreSQL or SQLite, or
- you want to learn logic programming and apply it to processing of Big Data.
-Among other engines, there is partial support for Trino and Postgres.
+Among other engines, there is partial support for Trino and Databricks.
Contributions to improve this support are
very welcome!
diff --git a/examples/more/Simple_Postgres.ipynb b/examples/more/Simple_Postgres.ipynb
new file mode 100644
index 00000000..4a96aec7
--- /dev/null
+++ b/examples/more/Simple_Postgres.ipynb
@@ -0,0 +1,138 @@
+{
+ "cells": [
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "id": "7e89b446",
+ "metadata": {},
+ "source": [
+ "# Example of running Postgres with text Pipeline overview"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "58f2b0db",
+ "metadata": {
+ "scrolled": true
+ },
+ "outputs": [],
+ "source": [
+ "from logica import colab_logica\n",
+ "colab_logica.ConnectToPostgres('interactive')\n",
+ "colab_logica.DISPLAY_MODE = 'colab-text'"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 4,
+ "id": "82c02eb6",
+ "metadata": {},
+ "outputs": [
+ {
+ "name": "stdout",
+ "output_type": "stream",
+ "text": [
+ "Query is stored at \u001b[1mGreeting_sql\u001b[0m variable.\n"
+ ]
+ },
+ {
+ "data": {
+ "text/html": [
+ ""
+ ],
+ "text/plain": [
+ ""
+ ]
+ },
+ "metadata": {},
+ "output_type": "display_data"
+ },
+ {
+ "name": "stdout",
+ "output_type": "stream",
+ "text": [
+ "Running predicate: Greeting (0 seconds)\n",
+ "The following table is stored at \u001b[1mGreeting\u001b[0m variable.\n"
+ ]
+ },
+ {
+ "data": {
+ "text/html": [
+ "\n",
+ "\n",
+ "
\n",
+ " \n",
+ " \n",
+ " | \n",
+ " col0 | \n",
+ "
\n",
+ " \n",
+ " \n",
+ " \n",
+ " | 0 | \n",
+ " Hello world! | \n",
+ "
\n",
+ " \n",
+ "
\n",
+ "