andygrove commented on code in PR #2748:
URL:
https://github.com/apache/datafusion-ballista/pull/2748#discussion_r4481568098
##########
ballista/core/src/object_store.rs:
##########
@@ -103,6 +104,7 @@ pub fn session_state_with_s3_support(
.with_runtime_env(runtime_env)
.with_config(session_config)
.with_default_features()
+
.with_optimizer_rules(crate::optimizer::ballista_default_optimizer_rules())
Review Comment:
This also reaches the Rust scheduler binary (`scheduler/src/bin/main.rs`
uses `session_state_with_s3_support`), so SQL planned on the scheduler, like
Flight SQL and Substrait submissions, now gets the NOT IN rewrite too. That's a
good fix! It'd just be worth a line in the PR description, since it currently
says there are no API changes and reads as Python only.
##########
python/python/ballista/jupyter.py:
##########
@@ -534,24 +533,36 @@ def _show_help(self) -> Optional[str]:
def explain_visual(df: DataFrame, analyze: bool = False) ->
ExecutionPlanVisualization:
"""
- Generate a visual representation of the execution plan.
+ Generate a visual representation of the query plan.
- This method creates an SVG visualization of the query execution plan,
- which can be displayed directly in Jupyter notebooks.
+ This method creates an SVG visualization of the DataFrame's logical plan,
+ which can be displayed directly in Jupyter notebooks. The query is not
+ executed.
Args:
df: DataFrame to visualize.
- analyze: If True, includes runtime statistics from actual execution.
+ analyze: Deprecated. It only labels the visualization; it does not
+ run the query or add runtime statistics, and passing ``True``
+ emits a ``DeprecationWarning``. Use ``df.explain(analyze=True)``
+ for runtime statistics.
Returns:
ExecutionPlanVisualization: An object that renders as SVG in Jupyter.
Example:
>>> df = ctx.sql("SELECT * FROM orders WHERE amount > 100")
>>> explain_visual(df) # Displays SVG in notebook
- >>> viz = explain_visual(df, analyze=True)
+ >>> viz = explain_visual(df)
>>> viz.save("plan.svg") # Save to file
"""
+ if analyze:
Review Comment:
Since `analyze=True` now emits a `DeprecationWarning`, could we mention the
deprecation in `docs/source/upgrading/56.0.0.md` as well?
##########
python/python/tests/test_context.py:
##########
@@ -302,6 +302,37 @@ def test_sql_explain_analyze(ctx):
assert "output_rows" in plan
+NOT_IN_QUERY = "SELECT a FROM t WHERE a NOT IN (SELECT b FROM s)"
+
+
+def register_not_in_tables(ctx, tmp_path):
+ pq.write_table(pa.table({"a": [1, 2, 3, 4, None]}), tmp_path / "t.parquet")
+ pq.write_table(pa.table({"b": [2, 4]}), tmp_path / "s.parquet")
+ ctx.register_parquet("t", str(tmp_path / "t.parquet"))
+ ctx.register_parquet("s", str(tmp_path / "s.parquet"))
+ ctx.sql("SET datafusion.optimizer.prefer_hash_join = true").collect()
+
+
+def test_not_in_subquery(ctx, tmp_path):
+ register_not_in_tables(ctx, tmp_path)
+ result = to_table(ctx.sql(NOT_IN_QUERY))
+ assert sorted(result.column("a").to_pylist()) == [1, 3]
+
+
[email protected](
Review Comment:
Could we file an issue for this gap and link it from the xfail reason? NOT
IN running as a single task through the Python client is worth tracking
somewhere people will see it, not only in a test.
##########
docs/source/user-guide/python/quickstart.md:
##########
@@ -182,6 +238,78 @@ assert result.column(1) == pyarrow.array([-3, -3, -3])
## User Defined Functions
-The underlying DataFusion query engine supports Python UDFs but this has not
yet been
-implemented in Ballista. It is planned for a future release. See
-[#173](https://github.com/apache/datafusion-ballista/issues/173) for status.
+Python scalar, aggregate and window UDFs are serialized with the plan using
+[cloudpickle](https://github.com/cloudpipe/cloudpickle), so they do not have
to be
+registered on the cluster:
+
+```python
+import pyarrow as pa
+from datafusion import col, udf
+
+
+def is_null(array: pa.Array) -> pa.Array:
+ return array.is_null()
+
+
+is_null_udf = udf(is_null, [pa.int64()], pa.bool_(), "stable",
name="is_null_udf")
+
+ctx.register_parquet("t", "./testdata/test.parquet")
+
+df = ctx.table("t")
+df.select(col("bigint_col"), is_null_udf(col("bigint_col")).alias("n")).show()
+
+# or with SQL
+ctx.register_udf(is_null_udf)
+ctx.sql("SELECT bigint_col, is_null_udf(bigint_col) AS n FROM t").show()
+```
+
+UDFs are run by Python on the executors, so the cluster has to meet these
requirements:
+
+- The scheduler and executors are started from Python, with
`BallistaScheduler`,
+ `BallistaExecutor` or `setup_test_cluster`, see below. The Rust
+ `ballista-scheduler` and `ballista-executor` binaries and the Docker images
can not
+ run Python UDFs.
+- They run the same Python minor version as the client.
+- Everything the UDF imports is installed on them.
+
+### Trust Model
+
+A cluster started from Python decodes UDFs with `cloudpickle.loads`, which can
run
+arbitrary Python code. Anyone who can reach the scheduler or executor ports
can therefore
+run arbitrary code on the cluster, as the user the processes run as. This is
the same trust
+model as PySpark.
+
+Only expose such a cluster to trusted clients, for example by binding to
`localhost` or a
+private network, or restricting access with a firewall.
+
+## Running a Cluster from Python
+
+The scheduler and executors can be started from Python. This is needed for
Python
+UDFs, and is convenient for development.
+
+```python
+from ballista import BallistaScheduler
+
+scheduler = BallistaScheduler(bind_host="0.0.0.0", bind_port=50050)
Review Comment:
Small nit: the trust model section just above suggests binding to
`localhost` or a private network, and then the examples bind to `0.0.0.0`.
Maybe use `localhost` here (and in the executor example), or add a short note
that `0.0.0.0` exposes the ports?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
