1 files changed, 18 insertions, 5 deletions
diff --git a/sqlglot/executor/__init__.py b/sqlglot/executor/__init__.py
index e765616..04621b5 100644
--- a/sqlglot/executor/__init__.py
+++ b/sqlglot/executor/__init__.py
@@ -1,20 +1,23 @@
 import logging
 import time
 
-from sqlglot import parse_one
+from sqlglot import maybe_parse
+from sqlglot.errors import ExecuteError
 from sqlglot.executor.python import PythonExecutor
+from sqlglot.executor.table import Table, ensure_tables
 from sqlglot.optimizer import optimize
 from sqlglot.planner import Plan
+from sqlglot.schema import ensure_schema
 
 logger = logging.getLogger("sqlglot")
 
 
-def execute(sql, schema, read=None):
+def execute(sql, schema=None, read=None, tables=None):
     """
     Run a sql query against data.
 
     Args:
-        sql (str): a sql statement
+        sql (str|sqlglot.Expression): a sql statement
         schema (dict|sqlglot.optimizer.Schema): database schema.
             This can either be an instance of `sqlglot.optimizer.Schema` or a mapping in one of
             the following forms:
@@ -23,10 +26,20 @@ def execute(sql, schema, read=None):
                 3. {catalog: {db: {table: {col: type}}}}
         read (str): the SQL dialect to apply during parsing
             (eg. "spark", "hive", "presto", "mysql").
+        tables (dict): additional tables to register.
     Returns:
         sqlglot.executor.Table: Simple columnar data structure.
     """
-    expression = parse_one(sql, read=read)
+    tables = ensure_tables(tables)
+    if not schema:
+        schema = {
+            name: {column: type(table[0][column]).__name__ for column in table.columns}
+            for name, table in tables.mapping.items()
+        }
+    schema = ensure_schema(schema)
+    if tables.supported_table_args and tables.supported_table_args != schema.supported_table_args:
+        raise ExecuteError("Tables must support the same table args as schema")
+    expression = maybe_parse(sql, dialect=read)
     now = time.time()
     expression = optimize(expression, schema, leave_tables_isolated=True)
     logger.debug("Optimization finished: %f", time.time() - now)
@@ -34,6 +47,6 @@ def execute(sql, schema, read=None):
     plan = Plan(expression)
     logger.debug("Logical Plan: %s", plan)
     now = time.time()
-    result = PythonExecutor().execute(plan)
+    result = PythonExecutor(tables=tables).execute(plan)
     logger.debug("Query finished: %f", time.time() - now)
     return result