Integrate SQLGlot for SQL to Pure Relation conversion

legendql/__init__.py

@@ -13,8 +13,8 @@
 # limitations under the License.
 
 from legendql.ql import db, table, using, using_db, using_db_def
-from legendql.ql import query, LegendQL
+from legendql.ql import query, LegendQL, from_sql
 from legendql.store import store
 from legendql.functions import aggregate, over, unbounded, rows, range, left, avg, count, sum, rank, lead, lag, row_number
 
-__all__ = ['db', 'table', 'using', 'using_db', 'using_db_def', 'query', 'LegendQL', 'store', 'aggregate', 'over', 'unbounded', 'rows', 'range', 'left', 'avg', 'count', 'sum', 'rank', 'lead', 'lag', 'row_number']
+__all__ = ['db', 'table', 'using', 'using_db', 'using_db_def', 'query', 'LegendQL', 'from_sql', 'store', 'aggregate', 'over', 'unbounded', 'rows', 'range', 'left', 'avg', 'count', 'sum', 'rank', 'lead', 'lag', 'row_number']

legendql/ql.py

@@ -61,6 +61,12 @@ def using_db(database: Database, table: str) -> LegendQL:
     return LegendQL.using_db_def(database, table, None)
 
 
+def from_sql(sql: str, database: Database) -> LegendQL:
+    """Create a LegendQL query from a SQL statement."""
+    query = Query.from_sql(sql, None, database)
+    return LegendQL.__new__(LegendQL)._initialize_with_query(query)
+
+
 class LegendQL:
 
     def __init__(self, database_definition: DatabaseDefinition, database: Database, table: str):
@@ -70,6 +76,11 @@ def __init__(self, database_definition: DatabaseDefinition, database: Database,
     def using_db_def(cls, database: Database, table: str, database_definition: DatabaseDefinition = None) -> LegendQL:
         return LegendQL(database_definition, database, table)
 
+    def _initialize_with_query(self, query: Query) -> LegendQL:
+        """Initialize LegendQL instance with an existing Query."""
+        self._query = query
+        return self
+
     def execute(self, store_or_def: Union[Store | DatabaseDefinition] = None, client: LegendClient = None) -> Result:
         database_def = None
         if store_or_def:

legendql/query.py

@@ -46,6 +46,33 @@ def from_table(cls, database_definition: DatabaseDefinition, database: Database,
         cloned_table = Table(found_table.table, found_table.columns.copy(), found_table.schema)
         return Query(database_definition, database, [cloned_table], cloned_table, [DatabaseFromClause(database.name, cloned_table.table, cloned_table.schema.name if cloned_table.schema else None)])
 
+    @classmethod
+    def from_sql(cls, sql: str, database_definition: DatabaseDefinition, database: Database) -> Query:
+        """Create a Query from a SQL statement."""
+        from legendql.sql_parser import SQLParser
+        from legendql.model.metamodel import DatabaseFromClause
+
+        parser = SQLParser(database)
+        clauses = parser.parse_sql(sql)
+
+        table_name = None
+        for clause in clauses:
+            if isinstance(clause, DatabaseFromClause):
+                table_name = clause.table
+                break
+
+        if not table_name:
+            raise ValueError("SQL statement must include a FROM clause with a table name")
+
+        all_tables = [x for xs in map(lambda c: [c] if isinstance(c, Table) else c, database.children) for x in xs]
+        try:
+            found_table = next(t for t in all_tables if t.table == table_name)
+        except StopIteration:
+            raise ValueError(f"Table {table_name} not found in the database.")
+
+        cloned_table = Table(found_table.table, found_table.columns.copy(), found_table.schema)
+        return Query(database_definition, database, [cloned_table], cloned_table, clauses)
+
     def execute(self, database_definition: DatabaseDefinition = None, client: LegendClient = None) -> Result:
         if database_definition is None and self._database_definition is None:
             raise ValueError("Database definition is required to execute the query.")

legendql/sql_parser.py

@@ -0,0 +1,213 @@
+# Copyright 2025 Goldman Sachs
+#
+# 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.
+
+from typing import List, Optional, Union
+import sqlglot
+from sqlglot import expressions as exp
+
+from legendql.model.metamodel import (
+    SelectionClause, FilterClause, LimitClause, OffsetClause, OrderByClause,
+    ColumnReferenceExpression, LiteralExpression, BinaryExpression,
+    IntegerLiteral, StringLiteral, BooleanLiteral, DateLiteral,
+    EqualsBinaryOperator, NotEqualsBinaryOperator, GreaterThanBinaryOperator,
+    LessThanBinaryOperator, GreaterThanEqualsBinaryOperator, LessThanEqualsBinaryOperator,
+    AndBinaryOperator, OrBinaryOperator, AddBinaryOperator, SubtractBinaryOperator,
+    MultiplyBinaryOperator, DivideBinaryOperator, OperandExpression,
+    OrderByExpression, AscendingOrderType, DescendingOrderType,
+    DatabaseFromClause, Clause, Expression
+)
+from legendql.model.schema import Database, Table
+import pyarrow as pa
+
+
+class SQLParser:
+    """Parser that converts SQL statements to LegendQL metamodel using SQLGlot."""
+
+    def __init__(self, database: Database):
+        self.database = database
+
+    def parse_sql(self, sql: str) -> List[Clause]:
+        """Parse SQL statement and return list of LegendQL clauses."""
+        try:
+            parsed = sqlglot.parse_one(sql)
+
+            if not isinstance(parsed, exp.Select):
+                raise ValueError(f"Only SELECT statements are currently supported, got: {type(parsed)}")
+
+            clauses = []
+
+            from_clause = self._parse_from_clause(parsed)
+            if from_clause:
+                clauses.append(from_clause)
+
+            select_clause = self._parse_select_clause(parsed)
+            if select_clause:
+                clauses.append(select_clause)
+
+            where_clause = self._parse_where_clause(parsed)
+            if where_clause:
+                clauses.append(where_clause)
+
+            order_by_clause = self._parse_order_by_clause(parsed)
+            if order_by_clause:
+                clauses.append(order_by_clause)
+
+            limit_clause = self._parse_limit_clause(parsed)
+            if limit_clause:
+                clauses.append(limit_clause)
+
+            offset_clause = self._parse_offset_clause(parsed)
+            if offset_clause:
+                clauses.append(offset_clause)
+
+            return clauses
+
+        except Exception as e:
+            raise ValueError(f"Failed to parse SQL: {sql}. Error: {str(e)}")
+
+    def _parse_from_clause(self, select_stmt: exp.Select) -> Optional[DatabaseFromClause]:
+        """Parse FROM clause and return DatabaseFromClause."""
+        if not select_stmt.find(exp.From):
+            return None
+
+        from_expr = select_stmt.find(exp.From)
+        table_expr = from_expr.find(exp.Table)
+
+        if not table_expr:
+            raise ValueError("FROM clause must specify a table")
+
+        table_name = table_expr.name
+        schema_name = table_expr.db if table_expr.db else None
+
+        database_name = self.database.name
+
+        return DatabaseFromClause(database_name, table_name, schema_name or "")
+
+    def _parse_select_clause(self, select_stmt: exp.Select) -> Optional[SelectionClause]:
+        """Parse SELECT clause and return SelectionClause."""
+        expressions = []
+
+        for select_expr in select_stmt.expressions:
+            if isinstance(select_expr, exp.Star):
+                raise ValueError("SELECT * is not currently supported. Please specify column names.")
+            elif isinstance(select_expr, exp.Column):
+                expressions.append(ColumnReferenceExpression(select_expr.name))
+            elif hasattr(select_expr, 'alias') and select_expr.alias:
+                expressions.append(ColumnReferenceExpression(select_expr.alias))
+            else:
+                expr = self._parse_expression(select_expr)
+                if isinstance(expr, ColumnReferenceExpression):
+                    expressions.append(expr)
+                else:
+                    if hasattr(select_expr, 'name'):
+                        expressions.append(ColumnReferenceExpression(select_expr.name))
+
+        return SelectionClause(expressions) if expressions else None
+
+    def _parse_where_clause(self, select_stmt: exp.Select) -> Optional[FilterClause]:
+        """Parse WHERE clause and return FilterClause."""
+        where_expr = select_stmt.find(exp.Where)
+        if not where_expr:
+            return None
+
+        condition = self._parse_expression(where_expr.this)
+        return FilterClause(condition)
+
+    def _parse_order_by_clause(self, select_stmt: exp.Select) -> Optional[OrderByClause]:
+        """Parse ORDER BY clause and return OrderByClause."""
+        order_by = select_stmt.find(exp.Order)
+        if not order_by:
+            return None
+
+        ordering = []
+        for ordered_expr in order_by.expressions:
+            column_expr = self._parse_expression(ordered_expr.this)
+            direction = DescendingOrderType() if ordered_expr.args.get('desc') else AscendingOrderType()
+            ordering.append(OrderByExpression(direction, column_expr))
+
+        return OrderByClause(ordering)
+
+    def _parse_limit_clause(self, select_stmt: exp.Select) -> Optional[LimitClause]:
+        """Parse LIMIT clause and return LimitClause."""
+        limit_expr = select_stmt.find(exp.Limit)
+        if not limit_expr:
+            return None
+
+        limit_value = self._parse_expression(limit_expr.expression)
+        if isinstance(limit_value, LiteralExpression) and isinstance(limit_value.literal, IntegerLiteral):
+            return LimitClause(limit_value.literal)
+
+        raise ValueError("LIMIT value must be an integer")
+
+    def _parse_offset_clause(self, select_stmt: exp.Select) -> Optional[OffsetClause]:
+        """Parse OFFSET clause and return OffsetClause."""
+        offset_expr = select_stmt.find(exp.Offset)
+        if not offset_expr:
+            return None
+
+        offset_value = self._parse_expression(offset_expr.expression)
+        if isinstance(offset_value, LiteralExpression) and isinstance(offset_value.literal, IntegerLiteral):
+            return OffsetClause(offset_value.literal)
+
+        raise ValueError("OFFSET value must be an integer")
+
+    def _parse_expression(self, expr) -> Expression:
+        """Parse SQLGlot expression and return LegendQL Expression."""
+        if isinstance(expr, exp.Column):
+            return ColumnReferenceExpression(expr.name)
+        elif isinstance(expr, exp.Literal):
+            return self._parse_literal(expr)
+        elif isinstance(expr, exp.Binary):
+            return self._parse_binary_expression(expr)
+        else:
+            if hasattr(expr, 'name'):
+                return ColumnReferenceExpression(expr.name)
+            raise ValueError(f"Unsupported expression type: {type(expr)}")
+
+    def _parse_literal(self, literal: exp.Literal) -> LiteralExpression:
+        """Parse SQLGlot literal and return LiteralExpression."""
+        if literal.is_int:
+            return LiteralExpression(IntegerLiteral(int(literal.this)))
+        elif literal.is_string:
+            return LiteralExpression(StringLiteral(literal.this))
+        else:
+            return LiteralExpression(StringLiteral(str(literal.this)))
+
+    def _parse_binary_expression(self, binary: exp.Binary) -> BinaryExpression:
+        """Parse SQLGlot binary expression and return BinaryExpression."""
+        left = OperandExpression(self._parse_expression(binary.left))
+        right = OperandExpression(self._parse_expression(binary.right))
+
+        operator_map = {
+            exp.EQ: EqualsBinaryOperator(),
+            exp.NEQ: NotEqualsBinaryOperator(),
+            exp.GT: GreaterThanBinaryOperator(),
+            exp.LT: LessThanBinaryOperator(),
+            exp.GTE: GreaterThanEqualsBinaryOperator(),
+            exp.LTE: LessThanEqualsBinaryOperator(),
+            exp.And: AndBinaryOperator(),
+            exp.Or: OrBinaryOperator(),
+            exp.Add: AddBinaryOperator(),
+            exp.Sub: SubtractBinaryOperator(),
+            exp.Mul: MultiplyBinaryOperator(),
+            exp.Div: DivideBinaryOperator(),
+        }
+
+        operator_class = type(binary)
+        if operator_class in operator_map:
+            operator = operator_map[operator_class]
+        else:
+            raise ValueError(f"Unsupported binary operator: {operator_class}")
+
+        return BinaryExpression(left, right, operator)

pyproject.toml

@@ -16,6 +16,7 @@ python = ">=3.9,<3.14"
 requests = ">=2.27.1"
 ijson = ">=3.1.4"
 pyarrow = ">=19.0.1"
+sqlglot = ">=25.0.0"
 pandas = [
     { version = ">=1.0.0", python = "<3.12" },
     { version = ">=2.1.1", python = ">=3.12" }

tests/legendql/sql/test_legendql_from_sql.py

@@ -0,0 +1,83 @@
+# Copyright 2025 Goldman Sachs
+#
+# 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 unittest
+import pyarrow as pa
+from legendql import from_sql
+from legendql.ql import LegendQL
+from legendql.model.schema import Database, Table
+
+
+class TestLegendQLFromSQL(unittest.TestCase):
+
+    def setUp(self):
+        """Set up test database."""
+        columns = [
+            pa.field("colA", pa.string()),
+            pa.field("colB", pa.int64()),
+            pa.field("colC", pa.float64())
+        ]
+        table = Table("tableC", columns)
+        self.database = Database("test_db", [table])
+
+    def test_from_sql_basic(self):
+        """Test creating LegendQL from basic SQL statement."""
+        sql = "SELECT colA, colB FROM tableC"
+        lql = from_sql(sql, self.database)
+
+        self.assertIsInstance(lql, LegendQL)
+        self.assertIsNotNone(lql._query)
+        self.assertEqual(len(lql._query._clauses), 2)
+
+    def test_from_sql_with_where(self):
+        """Test creating LegendQL from SQL statement with WHERE clause."""
+        sql = "SELECT colA FROM tableC WHERE colB = 42"
+        lql = from_sql(sql, self.database)
+
+        self.assertIsInstance(lql, LegendQL)
+        self.assertIsNotNone(lql._query)
+        self.assertEqual(len(lql._query._clauses), 3)
+
+    def test_from_sql_to_string(self):
+        """Test that LegendQL created from SQL can be converted to string."""
+        sql = "SELECT colA, colB FROM tableC"
+        lql = from_sql(sql, self.database)
+
+        result_string = lql.to_string()
+        self.assertIsInstance(result_string, str)
+        self.assertIn("tableC", result_string)
+        self.assertIn("colA", result_string)
+        self.assertIn("colB", result_string)
+
+    def test_from_sql_chaining(self):
+        """Test that LegendQL created from SQL supports method chaining."""
+        sql = "SELECT colA FROM tableC"
+        lql = from_sql(sql, self.database)
+
+        result = lql.limit(10)
+        self.assertIsInstance(result, LegendQL)
+        self.assertEqual(len(result._query._clauses), 3)  # FROM, SELECT, LIMIT
+
+    def test_from_sql_invalid_table(self):
+        """Test that from_sql raises error for non-existent table."""
+        sql = "SELECT colA FROM nonexistent_table"
+
+        with self.assertRaises(ValueError) as context:
+            from_sql(sql, self.database)
+
+        self.assertIn("Table nonexistent_table not found", str(context.exception))
+
+
+if __name__ == '__main__':
+    unittest.main()