You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "wengh (via GitHub)" <gi...@apache.org> on 2025/02/14 22:48:52 UTC
[PR] [WIP][PYTHON] Add filter pushdown API to Python Data Sources [spark]
wengh opened a new pull request, #49961:
URL: https://github.com/apache/spark/pull/49961
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1984839577
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
+ private var supportedFilters: Array[Filter] = Array.empty
+
+ private def metadata: Map[String, String] = {
+ Map(
+ "PushedFilters" -> supportedFilters.mkString("[", ", ", "]"),
+ "ReadSchema" -> outputSchema.simpleString
+ )
+ }
+
+ override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options, metadata)
+
+ // Optionally called by DSv2 once to push down filters before the scan is built.
+ override def pushFilters(filters: Array[Filter]): Array[Filter] = {
+ if (!SQLConf.get.pythonFilterPushDown) {
Review Comment:
It means we should always pushdown filters if `PythonScanBuilder` supports `SupportsPushDownFilters`.
Why do we need this config?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1972421058
##########
python/pyspark/sql/datasource.py:
##########
@@ -280,6 +306,9 @@ class DataSourceReader(ABC):
.. versionadded: 4.0.0
"""
+ def pushdownFilters(self, filters: List["Filter"]) -> Iterable["Filter"]:
Review Comment:
Does it follow the same naming convention as DSv2?
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -69,7 +70,7 @@ def latest_offset_func(reader: DataSourceStreamReader, outfile: IO) -> None:
def partitions_func(
reader: DataSourceStreamReader,
- data_source: DataSource,
+ data_source_name: str,
Review Comment:
why do we need this change?
##########
python/pyspark/sql/worker/data_source_pushdown_filters.py:
##########
@@ -0,0 +1,104 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 IO, List
+
+from pyspark.errors import PySparkAssertionError, PySparkValueError
+from pyspark.serializers import (
+ UTF8Deserializer,
+ read_int,
+ write_int,
+)
+from pyspark.sql.datasource import DataSourceReader, EqualTo, Filter
+from pyspark.sql.worker.internal.data_source_reader_info import DataSourceReaderInfo
+from pyspark.sql.worker.internal.data_source_worker import worker_main
+from pyspark.worker_util import (
+ read_command,
+ pickleSer,
+)
+
+utf8_deserializer = UTF8Deserializer()
+
+
+@worker_main
+def main(infile: IO, outfile: IO) -> None:
+ # Receive the data source reader instance.
+ reader_info = read_command(pickleSer, infile)
+ if not isinstance(reader_info, DataSourceReaderInfo):
+ raise PySparkAssertionError(
+ errorClass="DATA_SOURCE_TYPE_MISMATCH",
+ messageParameters={
+ "expected": "a Python data source reader info of type 'DataSourceReaderInfo'",
+ "actual": f"'{type(reader_info).__name__}'",
+ },
+ )
+
+ reader = reader_info.reader
+ if not isinstance(reader, DataSourceReader):
+ raise PySparkAssertionError(
+ errorClass="DATA_SOURCE_TYPE_MISMATCH",
+ messageParameters={
+ "expected": "a Python data source reader of type 'DataSourceReader'",
+ "actual": f"'{type(reader_info).__name__}'",
+ },
+ )
+
+ # Receive the pushdown filters.
+ num_filters = read_int(infile)
+ filters: List[Filter] = []
+ for _ in range(num_filters):
+ name = utf8_deserializer.loads(infile)
+ if name == "EqualTo":
Review Comment:
This seems fragile
##########
python/pyspark/sql/tests/test_python_datasource.py:
##########
@@ -246,6 +248,137 @@ def reader(self, schema) -> "DataSourceReader":
assertDataFrameEqual(df, [Row(x=0, y="0"), Row(x=1, y="1")])
self.assertEqual(df.select(spark_partition_id()).distinct().count(), 2)
+ def test_filter_pushdown(self):
+ class TestDataSourceReader(DataSourceReader):
+ def __init__(self):
+ self.has_filter = False
+
+ def pushdownFilters(self, filters: List[Filter]) -> Iterable[Filter]:
+ assert set(filters) == {
+ EqualTo(("x",), 1),
+ EqualTo(("y",), 2),
+ }, filters
+ self.has_filter = True
+ # pretend we support x = 1 filter but in fact we don't
+ # so we only return y = 2 filter
+ yield filters[filters.index(EqualTo(("y",), 2))]
+
+ def partitions(self):
+ assert self.has_filter
+ return super().partitions()
+
+ def read(self, partition):
+ assert self.has_filter
+ yield [1, 1]
+ yield [1, 2]
+ yield [2, 2]
+
+ class TestDataSource(DataSource):
+ @classmethod
+ def name(cls):
+ return "test"
+
+ def schema(self):
+ return "x int, y int"
+
+ def reader(self, schema) -> "DataSourceReader":
+ return TestDataSourceReader()
+
+ self.spark.dataSource.register(TestDataSource)
+ df = self.spark.read.format("test").load().filter("x = 1 and y = 2")
Review Comment:
I think we should also add Scala side test to check the query plan. Please see PythonDataSourceSuite.scala
##########
python/pyspark/sql/datasource.py:
##########
@@ -234,6 +249,35 @@ def streamReader(self, schema: StructType) -> "DataSourceStreamReader":
)
+ColumnPath = Tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class Filter(ABC):
+ """
+ The base class for filters used for filter pushdown.
+
+ .. versionadded: 4.1.0
+
Review Comment:
Let's also add some examples here
##########
python/pyspark/sql/datasource.py:
##########
@@ -234,6 +249,17 @@ def streamReader(self, schema: StructType) -> "DataSourceStreamReader":
)
+@dataclass(frozen=True)
+class Filter(ABC):
+ pass
Review Comment:
Can we add examples and docstring here? Let's also mention the limitations here (currently only support EqualTo)
##########
python/pyspark/sql/datasource.py:
##########
@@ -234,6 +249,35 @@ def streamReader(self, schema: StructType) -> "DataSourceStreamReader":
)
+ColumnPath = Tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class Filter(ABC):
+ """
+ The base class for filters used for filter pushdown.
+
+ .. versionadded: 4.1.0
+
+ Notes
+ -----
+ Column references are represented as a tuple of strings. For example, the column
+ `col1` is represented as `("col1",)`, and the nested column `a.b.c` is
+ represented as `("a", "b", "c")`.
+
+ Literal values are represented as Python objects of types such as
+ `int`, `float`, `str`, `bool`, `datetime`, etc.
+ See `Data Types <https://spark.apache.org/docs/latest/sql-ref-datatypes.html>`_
+ for more information about how values are represented in Python.
+ """
+
+
+@dataclass(frozen=True)
+class EqualTo(Filter):
+ lhsColumnPath: ColumnPath
+ rhsValue: Any
Review Comment:
Can we follow the DSv1 filter design here: https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/sources/filters.scala#L98
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +26,24 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
+ private var supported: Array[Filter] = Array.empty
+
override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+
+ override def pushFilters(filters: Array[Filter]): Array[Filter] = {
+ val dataSource = ds.getOrCreateDataSourceInPython(shortName, options, Some(outputSchema))
+ val result = ds.source.pushdownFiltersInPython(dataSource, outputSchema, filters)
+ ds.setDataSourceInPython(dataSource.copy(dataSource = result.dataSource))
Review Comment:
So here we are updating the serialized data source instance to get the new states stored in the data source instance. Can we add some comments here (e.g when will the pushedFilters be invoked during analysis, and why do we need to set the data source instance here)
##########
python/pyspark/sql/worker/internal/data_source_worker.py:
##########
@@ -0,0 +1,90 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 faulthandler
+from functools import wraps
+import os
+import sys
+from typing import IO, Callable
+
+from pyspark.accumulators import _accumulatorRegistry
+from pyspark.serializers import (
+ read_int,
+ write_int,
+ SpecialLengths,
+)
+from pyspark.util import handle_worker_exception, local_connect_and_auth
+from pyspark.worker_util import (
+ check_python_version,
+ send_accumulator_updates,
+ setup_broadcasts,
+ setup_memory_limits,
+ setup_spark_files,
+)
+
+
+F = Callable[[IO, IO], None]
+
+
+def worker_main(func: F) -> F:
Review Comment:
Let's avoid refactoring and cleaning up code in this PR :)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1972877821
##########
python/pyspark/sql/worker/data_source_pushdown_filters.py:
##########
@@ -0,0 +1,104 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 IO, List
+
+from pyspark.errors import PySparkAssertionError, PySparkValueError
+from pyspark.serializers import (
+ UTF8Deserializer,
+ read_int,
+ write_int,
+)
+from pyspark.sql.datasource import DataSourceReader, EqualTo, Filter
+from pyspark.sql.worker.internal.data_source_reader_info import DataSourceReaderInfo
+from pyspark.sql.worker.internal.data_source_worker import worker_main
+from pyspark.worker_util import (
+ read_command,
+ pickleSer,
+)
+
+utf8_deserializer = UTF8Deserializer()
+
+
+@worker_main
+def main(infile: IO, outfile: IO) -> None:
+ # Receive the data source reader instance.
+ reader_info = read_command(pickleSer, infile)
+ if not isinstance(reader_info, DataSourceReaderInfo):
+ raise PySparkAssertionError(
+ errorClass="DATA_SOURCE_TYPE_MISMATCH",
+ messageParameters={
+ "expected": "a Python data source reader info of type 'DataSourceReaderInfo'",
+ "actual": f"'{type(reader_info).__name__}'",
+ },
+ )
+
+ reader = reader_info.reader
+ if not isinstance(reader, DataSourceReader):
+ raise PySparkAssertionError(
+ errorClass="DATA_SOURCE_TYPE_MISMATCH",
+ messageParameters={
+ "expected": "a Python data source reader of type 'DataSourceReader'",
+ "actual": f"'{type(reader_info).__name__}'",
+ },
+ )
+
+ # Receive the pushdown filters.
+ num_filters = read_int(infile)
+ filters: List[Filter] = []
+ for _ in range(num_filters):
+ name = utf8_deserializer.loads(infile)
+ if name == "EqualTo":
Review Comment:
Although the filter name is still not going to be enforced at typecheck to match the Scala code. I guess this is the cost we pay for not using something like protobuf.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r2002148510
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScan.scala:
##########
@@ -16,26 +16,43 @@
*/
package org.apache.spark.sql.execution.datasources.v2.python
+import org.apache.commons.lang3.StringUtils
+
import org.apache.spark.JobArtifactSet
+import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.connector.metric.CustomMetric
import org.apache.spark.sql.connector.read._
import org.apache.spark.sql.connector.read.streaming.MicroBatchStream
+import org.apache.spark.sql.internal.connector.SupportsMetadata
+import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
-
+import org.apache.spark.util.Utils
class PythonScan(
- ds: PythonDataSourceV2,
- shortName: String,
- outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends Scan {
+ ds: PythonDataSourceV2,
+ shortName: String,
+ outputSchema: StructType,
+ options: CaseInsensitiveStringMap,
+ supportedFilters: Array[Filter]
+) extends Scan with SupportsMetadata {
+ private lazy val sparkSession = SparkSession.active
override def toBatch: Batch = new PythonBatch(ds, shortName, outputSchema, options)
override def toMicroBatchStream(checkpointLocation: String): MicroBatchStream =
new PythonMicroBatchStream(ds, shortName, outputSchema, options)
- override def description: String = "(Python)"
+ override def description: String = {
Review Comment:
Can we move this change into a separate PR?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1985895545
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
+ private var supportedFilters: Array[Filter] = Array.empty
+
+ private def metadata: Map[String, String] = {
+ Map(
+ "PushedFilters" -> supportedFilters.mkString("[", ", ", "]"),
+ "ReadSchema" -> outputSchema.simpleString
+ )
+ }
+
+ override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options, metadata)
+
+ // Optionally called by DSv2 once to push down filters before the scan is built.
+ override def pushFilters(filters: Array[Filter]): Array[Filter] = {
+ if (!SQLConf.get.pythonFilterPushDown) {
Review Comment:
Thank you for explanation.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r2008097077
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -4673,6 +4673,13 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val PYTHON_FILTER_PUSHDOWN_ENABLED = buildConf("spark.sql.python.filterPushdown.enabled")
+ .doc("When true, enable filter pushdown to Python datasource, at the cost of running " +
+ "Python worker one additional time during planning.")
Review Comment:
https://github.com/apache/spark/pull/50340
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1974559481
##########
python/pyspark/sql/tests/test_python_datasource.py:
##########
@@ -246,6 +248,137 @@ def reader(self, schema) -> "DataSourceReader":
assertDataFrameEqual(df, [Row(x=0, y="0"), Row(x=1, y="1")])
self.assertEqual(df.select(spark_partition_id()).distinct().count(), 2)
+ def test_filter_pushdown(self):
+ class TestDataSourceReader(DataSourceReader):
+ def __init__(self):
+ self.has_filter = False
+
+ def pushdownFilters(self, filters: List[Filter]) -> Iterable[Filter]:
+ assert set(filters) == {
+ EqualTo(("x",), 1),
+ EqualTo(("y",), 2),
+ }, filters
+ self.has_filter = True
+ # pretend we support x = 1 filter but in fact we don't
+ # so we only return y = 2 filter
+ yield filters[filters.index(EqualTo(("y",), 2))]
+
+ def partitions(self):
+ assert self.has_filter
+ return super().partitions()
+
+ def read(self, partition):
+ assert self.has_filter
+ yield [1, 1]
+ yield [1, 2]
+ yield [2, 2]
+
+ class TestDataSource(DataSource):
+ @classmethod
+ def name(cls):
+ return "test"
+
+ def schema(self):
+ return "x int, y int"
+
+ def reader(self, schema) -> "DataSourceReader":
+ return TestDataSourceReader()
+
+ self.spark.dataSource.register(TestDataSource)
+ df = self.spark.read.format("test").load().filter("x = 1 and y = 2")
Review Comment:
added pushed filters to the physical plan description and a scala test for the physical plan
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1984616581
##########
python/pyspark/errors/error-conditions.json:
##########
@@ -189,11 +189,21 @@
"Remote client cannot create a SparkContext. Create SparkSession instead."
]
},
+ "DATA_SOURCE_EXTRANEOUS_FILTERS": {
+ "message": [
+ "<type>.pushFilters() returned <extraneous> that are not part of the input <input>. Make sure that the returned filters are a subset of the input filters."
+ ]
+ },
"DATA_SOURCE_INVALID_RETURN_TYPE": {
"message": [
"Unsupported return type ('<type>') from Python data source '<name>'. Expected types: <supported_types>."
]
},
+ "DATA_SOURCE_PUSHDOWN_DISABLED": {
+ "message": [
+ "<type> implements pushFilters() but filter pushdown is disabled because '<conf>' is False. Run 'spark.conf.set(\"<conf>\", True)' to enable filter pushdown."
Review Comment:
`spark.conf.set(\"<conf>\", True)` is python syntax. We can just ask users to set the conf to true, without mentioning the concrete API.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1984615352
##########
python/pyspark/errors/error-conditions.json:
##########
@@ -189,11 +189,21 @@
"Remote client cannot create a SparkContext. Create SparkSession instead."
]
},
+ "DATA_SOURCE_EXTRANEOUS_FILTERS": {
+ "message": [
+ "<type>.pushFilters() returned <extraneous> that are not part of the input <input>. Make sure that the returned filters are a subset of the input filters."
Review Comment:
printing all the pushed filters might be too heavy for the error message. Maybe just mention the extraneous filters?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1984636973
##########
python/pyspark/sql/datasource.py:
##########
@@ -234,6 +249,62 @@ def streamReader(self, schema: StructType) -> "DataSourceStreamReader":
)
+ColumnPath = Tuple[str, ...]
+
+
+@dataclass(frozen=True)
Review Comment:
means that it's immutable
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r2002750595
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -4673,6 +4673,13 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val PYTHON_FILTER_PUSHDOWN_ENABLED = buildConf("spark.sql.python.filterPushdown.enabled")
+ .doc("When true, enable filter pushdown to Python datasource, at the cost of running " +
+ "Python worker one additional time during planning.")
Review Comment:
do we still have additional planning now?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1972843589
##########
python/pyspark/sql/datasource.py:
##########
@@ -234,6 +249,35 @@ def streamReader(self, schema: StructType) -> "DataSourceStreamReader":
)
+ColumnPath = Tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class Filter(ABC):
+ """
+ The base class for filters used for filter pushdown.
+
+ .. versionadded: 4.1.0
+
+ Notes
+ -----
+ Column references are represented as a tuple of strings. For example, the column
+ `col1` is represented as `("col1",)`, and the nested column `a.b.c` is
+ represented as `("a", "b", "c")`.
+
+ Literal values are represented as Python objects of types such as
+ `int`, `float`, `str`, `bool`, `datetime`, etc.
+ See `Data Types <https://spark.apache.org/docs/latest/sql-ref-datatypes.html>`_
+ for more information about how values are represented in Python.
+ """
+
+
+@dataclass(frozen=True)
+class EqualTo(Filter):
+ lhsColumnPath: ColumnPath
+ rhsValue: Any
Review Comment:
Yeah we can rename the fields to `attribute` and `value` but still using `Tuple[str, ...]` for attribute rather than `str`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1972838057
##########
python/pyspark/sql/datasource.py:
##########
@@ -280,6 +306,9 @@ class DataSourceReader(ABC):
.. versionadded: 4.0.0
"""
+ def pushdownFilters(self, filters: List["Filter"]) -> Iterable["Filter"]:
Review Comment:
It should be `pushFilters`.
https://github.com/apache/spark/blob/727167acc30c7a50566dad0c030763e34b450cca/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownFilters.java#L30-L38
##########
python/pyspark/sql/datasource.py:
##########
@@ -234,6 +249,17 @@ def streamReader(self, schema: StructType) -> "DataSourceStreamReader":
)
+@dataclass(frozen=True)
+class Filter(ABC):
+ pass
Review Comment:
Added
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1972878704
##########
python/pyspark/sql/datasource.py:
##########
@@ -280,6 +306,9 @@ class DataSourceReader(ABC):
.. versionadded: 4.0.0
"""
+ def pushdownFilters(self, filters: List["Filter"]) -> Iterable["Filter"]:
Review Comment:
updated
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1984618409
##########
python/pyspark/sql/datasource.py:
##########
@@ -234,6 +249,62 @@ def streamReader(self, schema: StructType) -> "DataSourceStreamReader":
)
+ColumnPath = Tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class Filter(ABC):
+ """
+ The base class for filters used for filter pushdown.
+
+ .. versionadded: 4.1.0
+
+ Notes
+ -----
+ Column references are represented as a tuple of strings. For example:
+
+ +----------------+----------------------+
+ | Column | Representation |
+ +----------------+----------------------+
+ | `col1` | `("col1",)` |
Review Comment:
```suggestion
| `col1` | `("col1")` |
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1974189473
##########
python/pyspark/sql/tests/test_python_datasource.py:
##########
@@ -246,6 +248,137 @@ def reader(self, schema) -> "DataSourceReader":
assertDataFrameEqual(df, [Row(x=0, y="0"), Row(x=1, y="1")])
self.assertEqual(df.select(spark_partition_id()).distinct().count(), 2)
+ def test_filter_pushdown(self):
+ class TestDataSourceReader(DataSourceReader):
+ def __init__(self):
+ self.has_filter = False
+
+ def pushdownFilters(self, filters: List[Filter]) -> Iterable[Filter]:
+ assert set(filters) == {
+ EqualTo(("x",), 1),
+ EqualTo(("y",), 2),
+ }, filters
+ self.has_filter = True
+ # pretend we support x = 1 filter but in fact we don't
+ # so we only return y = 2 filter
+ yield filters[filters.index(EqualTo(("y",), 2))]
+
+ def partitions(self):
+ assert self.has_filter
+ return super().partitions()
+
+ def read(self, partition):
+ assert self.has_filter
+ yield [1, 1]
+ yield [1, 2]
+ yield [2, 2]
+
+ class TestDataSource(DataSource):
+ @classmethod
+ def name(cls):
+ return "test"
+
+ def schema(self):
+ return "x int, y int"
+
+ def reader(self, schema) -> "DataSourceReader":
+ return TestDataSourceReader()
+
+ self.spark.dataSource.register(TestDataSource)
+ df = self.spark.read.format("test").load().filter("x = 1 and y = 2")
Review Comment:
like this
```
*(1) Project [x#296, y#297]
+- *(1) Filter ((isnotnull(x#296) AND isnotnull(y#297)) AND (y#297 = 2))
+- BatchScan test[x#296, y#297] (Python) PushedFilters: [EqualTo(x,1)], ReadSchema: struct<x:int,y:int>, ShortName: test RuntimeFilters: []
```
##########
python/pyspark/sql/tests/test_python_datasource.py:
##########
@@ -246,6 +248,137 @@ def reader(self, schema) -> "DataSourceReader":
assertDataFrameEqual(df, [Row(x=0, y="0"), Row(x=1, y="1")])
self.assertEqual(df.select(spark_partition_id()).distinct().count(), 2)
+ def test_filter_pushdown(self):
+ class TestDataSourceReader(DataSourceReader):
+ def __init__(self):
+ self.has_filter = False
+
+ def pushdownFilters(self, filters: List[Filter]) -> Iterable[Filter]:
+ assert set(filters) == {
+ EqualTo(("x",), 1),
+ EqualTo(("y",), 2),
+ }, filters
+ self.has_filter = True
+ # pretend we support x = 1 filter but in fact we don't
+ # so we only return y = 2 filter
+ yield filters[filters.index(EqualTo(("y",), 2))]
+
+ def partitions(self):
+ assert self.has_filter
+ return super().partitions()
+
+ def read(self, partition):
+ assert self.has_filter
+ yield [1, 1]
+ yield [1, 2]
+ yield [2, 2]
+
+ class TestDataSource(DataSource):
+ @classmethod
+ def name(cls):
+ return "test"
+
+ def schema(self):
+ return "x int, y int"
+
+ def reader(self, schema) -> "DataSourceReader":
+ return TestDataSourceReader()
+
+ self.spark.dataSource.register(TestDataSource)
+ df = self.spark.read.format("test").load().filter("x = 1 and y = 2")
Review Comment:
like this
```
*(1) Project [x#296, y#297]
+- *(1) Filter ((isnotnull(x#296) AND isnotnull(y#297)) AND (y#297 = 2))
+- BatchScan test[x#296, y#297] (Python) PushedFilters: [EqualTo(x,1)], ReadSchema: struct<x:int,y:int> RuntimeFilters: []
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r2008102653
##########
python/pyspark/sql/worker/data_source_pushdown_filters.py:
##########
@@ -0,0 +1,104 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 IO, List
+
+from pyspark.errors import PySparkAssertionError, PySparkValueError
+from pyspark.serializers import (
+ UTF8Deserializer,
+ read_int,
+ write_int,
+)
+from pyspark.sql.datasource import DataSourceReader, EqualTo, Filter
+from pyspark.sql.worker.internal.data_source_reader_info import DataSourceReaderInfo
+from pyspark.sql.worker.internal.data_source_worker import worker_main
+from pyspark.worker_util import (
+ read_command,
+ pickleSer,
+)
+
+utf8_deserializer = UTF8Deserializer()
+
+
+@worker_main
+def main(infile: IO, outfile: IO) -> None:
+ # Receive the data source reader instance.
+ reader_info = read_command(pickleSer, infile)
+ if not isinstance(reader_info, DataSourceReaderInfo):
+ raise PySparkAssertionError(
+ errorClass="DATA_SOURCE_TYPE_MISMATCH",
+ messageParameters={
+ "expected": "a Python data source reader info of type 'DataSourceReaderInfo'",
+ "actual": f"'{type(reader_info).__name__}'",
+ },
+ )
+
+ reader = reader_info.reader
+ if not isinstance(reader, DataSourceReader):
+ raise PySparkAssertionError(
+ errorClass="DATA_SOURCE_TYPE_MISMATCH",
+ messageParameters={
+ "expected": "a Python data source reader of type 'DataSourceReader'",
+ "actual": f"'{type(reader_info).__name__}'",
+ },
+ )
+
+ # Receive the pushdown filters.
+ num_filters = read_int(infile)
+ filters: List[Filter] = []
+ for _ in range(num_filters):
+ name = utf8_deserializer.loads(infile)
+ if name == "EqualTo":
Review Comment:
https://github.com/apache/spark/pull/50252
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1978530686
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
Review Comment:
cc @cloud-fan
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on PR #49961:
URL: https://github.com/apache/spark/pull/49961#issuecomment-2738739311
Thanks, merging to master.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1984790606
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/UserDefinedPythonDataSource.scala:
##########
@@ -300,6 +321,94 @@ private class UserDefinedPythonDataSourceRunner(
}
}
+case class PythonFilterPushdownResult(
+ dataSource: Array[Byte],
+ isFilterPushed: collection.Seq[Boolean])
Review Comment:
The name `isFilterPushed` looks confused. In fact, it contains both pushed and not pushed.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1985896731
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
+ private var supportedFilters: Array[Filter] = Array.empty
+
+ private def metadata: Map[String, String] = {
+ Map(
+ "PushedFilters" -> supportedFilters.mkString("[", ", ", "]"),
+ "ReadSchema" -> outputSchema.simpleString
+ )
+ }
+
+ override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options, metadata)
+
+ // Optionally called by DSv2 once to push down filters before the scan is built.
+ override def pushFilters(filters: Array[Filter]): Array[Filter] = {
+ if (!SQLConf.get.pythonFilterPushDown) {
+ return filters
+ }
+
+ val dataSource = ds.getOrCreateDataSourceInPython(shortName, options, Some(outputSchema))
+ val result = ds.source.pushdownFiltersInPython(dataSource, outputSchema, filters)
+
+ // The Data Source instance state changes after pushdown to remember the reader instance
+ // created and the filters pushed down. So pushdownFiltersInPython returns a new pickled
+ // Data Source instance. We need to use that new instance for further operations.
+ ds.setDataSourceInPython(dataSource.copy(dataSource = result.dataSource))
+
+ // Partition the filters into supported and unsupported ones.
+ val isPushed = result.isFilterPushed.zip(filters)
+ supportedFilters = isPushed.collect { case (true, filter) => filter }.toArray
+ val unsupported = isPushed.collect { case (false, filter) => filter }.toArray
+ unsupported
Review Comment:
~~I think we should keep all the filters even if part of them can be pushed.~~
~~This will ensure the robust if the python data source works incorrect.~~
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r2002218436
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonDataSourceSuite.scala:
##########
@@ -213,6 +220,66 @@ class PythonDataSourceSuite extends PythonDataSourceSuiteBase {
parameters = Map("inputSchema" -> "INT", "dataType" -> "\"INT\""))
}
+ test("data source reader with filter pushdown") {
+ assume(shouldTestPandasUDFs)
+ val dataSourceScript =
+ s"""
+ |from pyspark.sql.datasource import (
+ | DataSource,
+ | DataSourceReader,
+ | EqualTo,
+ | InputPartition,
+ |)
+ |
+ |class SimpleDataSourceReader(DataSourceReader):
+ | def partitions(self):
+ | return [InputPartition(i) for i in range(2)]
+ |
+ | def pushFilters(self, filters):
+ | yield filters[filters.index(EqualTo(("id",), 1))]
+ |
+ | def read(self, partition):
+ | yield (0, partition.value)
+ | yield (1, partition.value)
+ | yield (2, partition.value)
+ |
+ |class SimpleDataSource(DataSource):
+ | def schema(self):
+ | return "id int, partition int"
+ |
+ | def reader(self, schema):
+ | return SimpleDataSourceReader()
+ |""".stripMargin
+ val schema = StructType.fromDDL("id INT, partition INT")
+ val dataSource =
+ createUserDefinedPythonDataSource(name = dataSourceName, pythonScript = dataSourceScript)
+ spark.conf.set(SQLConf.PYTHON_FILTER_PUSHDOWN_ENABLED, true)
Review Comment:
> Can we also add another test case with this config disabled?
We do have Python tests with the config disabled. Is that ok?
The reason we have a Scala test was to check the physical plan which is not really possible in Python.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1972861639
##########
python/pyspark/sql/datasource.py:
##########
@@ -234,6 +249,35 @@ def streamReader(self, schema: StructType) -> "DataSourceStreamReader":
)
+ColumnPath = Tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class Filter(ABC):
+ """
+ The base class for filters used for filter pushdown.
+
+ .. versionadded: 4.1.0
+
Review Comment:
Should add examples of sql filter expression -> filter and also unsupported filter examples
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on PR #49961:
URL: https://github.com/apache/spark/pull/49961#issuecomment-2719735902
cc @cloud-fan @allisonwang-db
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1984418894
##########
python/pyspark/sql/datasource.py:
##########
@@ -234,6 +249,62 @@ def streamReader(self, schema: StructType) -> "DataSourceStreamReader":
)
+ColumnPath = Tuple[str, ...]
Review Comment:
add comment to explain
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #49961:
URL: https://github.com/apache/spark/pull/49961#issuecomment-2705841733
For DS v2, the scan workflow is:
1. analyzer gets `Table` from `TableProvider`, and puts it in `DataSourceV2Relation` (for batch scan) or `StreamingRelationV2` (for streaming scan). The `Table` reports schema because the relation logical plan needs it.
2. optimizer gets `Scan` from `Table` by doing operator pushdown, and turns the relation into `DataSourceV2ScanRelation`. (streaming does not support filter pushdown yet)
3. planner gets `Batch` or `Stream` from `Scan` to build the physical scan node.
Python data source API is a bit different from DS v2 and there is no 1-1 mapping. I think the current mapping is
1. analyzer gets pickled python `DataSource` instance and keeps it in `PythonDataSourceV2` which extends `TableProvider`. The `PythonTable` (extends `Table`) simply references `PythonDataSourceV2`.
2. optimizer gets `PythonScan` (extends `Scan`) from `PythonTable`. `PythonScan` also simply references `PythonTable` and does nothing else.
3. planner gets `PythonBatch` (extends `Batch`) from `PythonScan`, which does some real job: launch a python worker to create the python batch reader and pickle it. The pickled instance is kept in `PythonBatch`.
Now to push down filters, we need to create the python batch reader earlier, which means one more round of Python worker communication in the optimizer. I'm wondering that once we finish pushdown, shall we do the planning work immediately and keep `PythonDataSourceReadInfo` in `PythonScan`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1985896731
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
+ private var supportedFilters: Array[Filter] = Array.empty
+
+ private def metadata: Map[String, String] = {
+ Map(
+ "PushedFilters" -> supportedFilters.mkString("[", ", ", "]"),
+ "ReadSchema" -> outputSchema.simpleString
+ )
+ }
+
+ override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options, metadata)
+
+ // Optionally called by DSv2 once to push down filters before the scan is built.
+ override def pushFilters(filters: Array[Filter]): Array[Filter] = {
+ if (!SQLConf.get.pythonFilterPushDown) {
+ return filters
+ }
+
+ val dataSource = ds.getOrCreateDataSourceInPython(shortName, options, Some(outputSchema))
+ val result = ds.source.pushdownFiltersInPython(dataSource, outputSchema, filters)
+
+ // The Data Source instance state changes after pushdown to remember the reader instance
+ // created and the filters pushed down. So pushdownFiltersInPython returns a new pickled
+ // Data Source instance. We need to use that new instance for further operations.
+ ds.setDataSourceInPython(dataSource.copy(dataSource = result.dataSource))
+
+ // Partition the filters into supported and unsupported ones.
+ val isPushed = result.isFilterPushed.zip(filters)
+ supportedFilters = isPushed.collect { case (true, filter) => filter }.toArray
+ val unsupported = isPushed.collect { case (false, filter) => filter }.toArray
+ unsupported
Review Comment:
I think we should keep all the filters even if part of them can be pushed.
This will ensure the robust if the python data source works incorrect.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db closed pull request #49961: [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources
URL: https://github.com/apache/spark/pull/49961
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1978082787
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
Review Comment:
We decided to only support V1 filters to keep the Python API simple.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1985905297
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScan.scala:
##########
@@ -16,26 +16,43 @@
*/
package org.apache.spark.sql.execution.datasources.v2.python
+import org.apache.commons.lang3.StringUtils
+
import org.apache.spark.JobArtifactSet
+import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.connector.metric.CustomMetric
import org.apache.spark.sql.connector.read._
import org.apache.spark.sql.connector.read.streaming.MicroBatchStream
+import org.apache.spark.sql.internal.connector.SupportsMetadata
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
-
+import org.apache.spark.util.Utils
class PythonScan(
- ds: PythonDataSourceV2,
- shortName: String,
- outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends Scan {
+ ds: PythonDataSourceV2,
+ shortName: String,
+ outputSchema: StructType,
+ options: CaseInsensitiveStringMap,
+ metadata: => Map[String, String]
+) extends Scan
+ with SupportsMetadata {
+ private lazy val sparkSession = SparkSession.active
+ private def maxMetadataValueLength = sparkSession.sessionState.conf.maxMetadataStringLength
Review Comment:
`maxMetadataValueLength` used only once.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1985505029
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
+ private var supportedFilters: Array[Filter] = Array.empty
+
+ private def metadata: Map[String, String] = {
+ Map(
+ "PushedFilters" -> supportedFilters.mkString("[", ", ", "]"),
+ "ReadSchema" -> outputSchema.simpleString
+ )
+ }
+
+ override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options, metadata)
+
+ // Optionally called by DSv2 once to push down filters before the scan is built.
+ override def pushFilters(filters: Array[Filter]): Array[Filter] = {
+ if (!SQLConf.get.pythonFilterPushDown) {
+ return filters
+ }
+
+ val dataSource = ds.getOrCreateDataSourceInPython(shortName, options, Some(outputSchema))
+ val result = ds.source.pushdownFiltersInPython(dataSource, outputSchema, filters)
+
+ // The Data Source instance state changes after pushdown to remember the reader instance
+ // created and the filters pushed down. So pushdownFiltersInPython returns a new pickled
+ // Data Source instance. We need to use that new instance for further operations.
+ ds.setDataSourceInPython(dataSource.copy(dataSource = result.dataSource))
+
+ // Partition the filters into supported and unsupported ones.
+ val isPushed = result.isFilterPushed.zip(filters)
+ supportedFilters = isPushed.collect { case (true, filter) => filter }.toArray
+ val unsupported = isPushed.collect { case (false, filter) => filter }.toArray
+ unsupported
Review Comment:
What do you mean? supportedFilters ++ unsupported is all filters. We keep supportedFilters to implement pushedFilters required by the interface.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1985560600
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/UserDefinedPythonDataSource.scala:
##########
@@ -300,6 +321,94 @@ private class UserDefinedPythonDataSourceRunner(
}
}
+case class PythonFilterPushdownResult(
+ dataSource: Array[Byte],
+ isFilterPushed: collection.Seq[Boolean])
Review Comment:
or add a comment to explain
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1992622771
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonDataSourceV2.scala:
##########
@@ -52,6 +52,11 @@ class PythonDataSourceV2 extends TableProvider {
dataSourceInPython
}
+ def setDataSourceInPython(
+ dataSourceInPython: PythonDataSourceCreationResult): Unit = {
Review Comment:
```suggestion
def setDataSourceInPython(dataSourceInPython: PythonDataSourceCreationResult): Unit = {
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1984638659
##########
python/pyspark/errors/error-conditions.json:
##########
@@ -189,11 +189,21 @@
"Remote client cannot create a SparkContext. Create SparkSession instead."
]
},
+ "DATA_SOURCE_EXTRANEOUS_FILTERS": {
+ "message": [
+ "<type>.pushFilters() returned <extraneous> that are not part of the input <input>. Make sure that the returned filters are a subset of the input filters."
Review Comment:
do the error parameters still show as part of the error even if the message format string doesn't use it? if so then we don't need to include it here yeah.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1972838920
##########
python/pyspark/sql/worker/internal/data_source_worker.py:
##########
@@ -0,0 +1,90 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 faulthandler
+from functools import wraps
+import os
+import sys
+from typing import IO, Callable
+
+from pyspark.accumulators import _accumulatorRegistry
+from pyspark.serializers import (
+ read_int,
+ write_int,
+ SpecialLengths,
+)
+from pyspark.util import handle_worker_exception, local_connect_and_auth
+from pyspark.worker_util import (
+ check_python_version,
+ send_accumulator_updates,
+ setup_broadcasts,
+ setup_memory_limits,
+ setup_spark_files,
+)
+
+
+F = Callable[[IO, IO], None]
+
+
+def worker_main(func: F) -> F:
Review Comment:
ok
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1972877821
##########
python/pyspark/sql/worker/data_source_pushdown_filters.py:
##########
@@ -0,0 +1,104 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 IO, List
+
+from pyspark.errors import PySparkAssertionError, PySparkValueError
+from pyspark.serializers import (
+ UTF8Deserializer,
+ read_int,
+ write_int,
+)
+from pyspark.sql.datasource import DataSourceReader, EqualTo, Filter
+from pyspark.sql.worker.internal.data_source_reader_info import DataSourceReaderInfo
+from pyspark.sql.worker.internal.data_source_worker import worker_main
+from pyspark.worker_util import (
+ read_command,
+ pickleSer,
+)
+
+utf8_deserializer = UTF8Deserializer()
+
+
+@worker_main
+def main(infile: IO, outfile: IO) -> None:
+ # Receive the data source reader instance.
+ reader_info = read_command(pickleSer, infile)
+ if not isinstance(reader_info, DataSourceReaderInfo):
+ raise PySparkAssertionError(
+ errorClass="DATA_SOURCE_TYPE_MISMATCH",
+ messageParameters={
+ "expected": "a Python data source reader info of type 'DataSourceReaderInfo'",
+ "actual": f"'{type(reader_info).__name__}'",
+ },
+ )
+
+ reader = reader_info.reader
+ if not isinstance(reader, DataSourceReader):
+ raise PySparkAssertionError(
+ errorClass="DATA_SOURCE_TYPE_MISMATCH",
+ messageParameters={
+ "expected": "a Python data source reader of type 'DataSourceReader'",
+ "actual": f"'{type(reader_info).__name__}'",
+ },
+ )
+
+ # Receive the pushdown filters.
+ num_filters = read_int(infile)
+ filters: List[Filter] = []
+ for _ in range(num_filters):
+ name = utf8_deserializer.loads(infile)
+ if name == "EqualTo":
Review Comment:
Although the filter name is still not going to be enforced at compile time to match the Scala code. I guess this is the cost we pay for not using something like protobuf.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1972877131
##########
python/pyspark/sql/worker/data_source_pushdown_filters.py:
##########
@@ -0,0 +1,104 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 IO, List
+
+from pyspark.errors import PySparkAssertionError, PySparkValueError
+from pyspark.serializers import (
+ UTF8Deserializer,
+ read_int,
+ write_int,
+)
+from pyspark.sql.datasource import DataSourceReader, EqualTo, Filter
+from pyspark.sql.worker.internal.data_source_reader_info import DataSourceReaderInfo
+from pyspark.sql.worker.internal.data_source_worker import worker_main
+from pyspark.worker_util import (
+ read_command,
+ pickleSer,
+)
+
+utf8_deserializer = UTF8Deserializer()
+
+
+@worker_main
+def main(infile: IO, outfile: IO) -> None:
+ # Receive the data source reader instance.
+ reader_info = read_command(pickleSer, infile)
+ if not isinstance(reader_info, DataSourceReaderInfo):
+ raise PySparkAssertionError(
+ errorClass="DATA_SOURCE_TYPE_MISMATCH",
+ messageParameters={
+ "expected": "a Python data source reader info of type 'DataSourceReaderInfo'",
+ "actual": f"'{type(reader_info).__name__}'",
+ },
+ )
+
+ reader = reader_info.reader
+ if not isinstance(reader, DataSourceReader):
+ raise PySparkAssertionError(
+ errorClass="DATA_SOURCE_TYPE_MISMATCH",
+ messageParameters={
+ "expected": "a Python data source reader of type 'DataSourceReader'",
+ "actual": f"'{type(reader_info).__name__}'",
+ },
+ )
+
+ # Receive the pushdown filters.
+ num_filters = read_int(infile)
+ filters: List[Filter] = []
+ for _ in range(num_filters):
+ name = utf8_deserializer.loads(infile)
+ if name == "EqualTo":
Review Comment:
This is temporary and will be replaced by JSON
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1974559481
##########
python/pyspark/sql/tests/test_python_datasource.py:
##########
@@ -246,6 +248,137 @@ def reader(self, schema) -> "DataSourceReader":
assertDataFrameEqual(df, [Row(x=0, y="0"), Row(x=1, y="1")])
self.assertEqual(df.select(spark_partition_id()).distinct().count(), 2)
+ def test_filter_pushdown(self):
+ class TestDataSourceReader(DataSourceReader):
+ def __init__(self):
+ self.has_filter = False
+
+ def pushdownFilters(self, filters: List[Filter]) -> Iterable[Filter]:
+ assert set(filters) == {
+ EqualTo(("x",), 1),
+ EqualTo(("y",), 2),
+ }, filters
+ self.has_filter = True
+ # pretend we support x = 1 filter but in fact we don't
+ # so we only return y = 2 filter
+ yield filters[filters.index(EqualTo(("y",), 2))]
+
+ def partitions(self):
+ assert self.has_filter
+ return super().partitions()
+
+ def read(self, partition):
+ assert self.has_filter
+ yield [1, 1]
+ yield [1, 2]
+ yield [2, 2]
+
+ class TestDataSource(DataSource):
+ @classmethod
+ def name(cls):
+ return "test"
+
+ def schema(self):
+ return "x int, y int"
+
+ def reader(self, schema) -> "DataSourceReader":
+ return TestDataSourceReader()
+
+ self.spark.dataSource.register(TestDataSource)
+ df = self.spark.read.format("test").load().filter("x = 1 and y = 2")
Review Comment:
added the description and a scala test for this
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1990448785
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScan.scala:
##########
@@ -16,26 +16,44 @@
*/
package org.apache.spark.sql.execution.datasources.v2.python
+import org.apache.commons.lang3.StringUtils
+
import org.apache.spark.JobArtifactSet
+import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.connector.metric.CustomMetric
import org.apache.spark.sql.connector.read._
import org.apache.spark.sql.connector.read.streaming.MicroBatchStream
+import org.apache.spark.sql.internal.connector.SupportsMetadata
+import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
-
+import org.apache.spark.util.Utils
class PythonScan(
- ds: PythonDataSourceV2,
- shortName: String,
- outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends Scan {
+ ds: PythonDataSourceV2,
+ shortName: String,
+ outputSchema: StructType,
+ options: CaseInsensitiveStringMap,
+ supportedFilters: Array[Filter]
+) extends Scan
+ with SupportsMetadata {
+ private lazy val sparkSession = SparkSession.active
override def toBatch: Batch = new PythonBatch(ds, shortName, outputSchema, options)
override def toMicroBatchStream(checkpointLocation: String): MicroBatchStream =
new PythonMicroBatchStream(ds, shortName, outputSchema, options)
- override def description: String = "(Python)"
+ override def description: String = {
+ val metadataStr = getMetaData().toSeq.sorted.map {
+ case (key, value) =>
+ val maxMetadataValueLength = sparkSession.sessionState.conf.maxMetadataStringLength
+ val redactedValue =
+ Utils.redact(sparkSession.sessionState.conf.stringRedactionPattern, value)
+ key + ": " + StringUtils.abbreviate(redactedValue, maxMetadataValueLength)
Review Comment:
```suggestion
val redactedValue =
Utils.redact(sparkSession.sessionState.conf.stringRedactionPattern, value)
key + ": " + StringUtils.abbreviate(
redactedValue, sparkSession.sessionState.conf.maxMetadataStringLength)
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1985505029
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
+ private var supportedFilters: Array[Filter] = Array.empty
+
+ private def metadata: Map[String, String] = {
+ Map(
+ "PushedFilters" -> supportedFilters.mkString("[", ", ", "]"),
+ "ReadSchema" -> outputSchema.simpleString
+ )
+ }
+
+ override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options, metadata)
+
+ // Optionally called by DSv2 once to push down filters before the scan is built.
+ override def pushFilters(filters: Array[Filter]): Array[Filter] = {
+ if (!SQLConf.get.pythonFilterPushDown) {
+ return filters
+ }
+
+ val dataSource = ds.getOrCreateDataSourceInPython(shortName, options, Some(outputSchema))
+ val result = ds.source.pushdownFiltersInPython(dataSource, outputSchema, filters)
+
+ // The Data Source instance state changes after pushdown to remember the reader instance
+ // created and the filters pushed down. So pushdownFiltersInPython returns a new pickled
+ // Data Source instance. We need to use that new instance for further operations.
+ ds.setDataSourceInPython(dataSource.copy(dataSource = result.dataSource))
+
+ // Partition the filters into supported and unsupported ones.
+ val isPushed = result.isFilterPushed.zip(filters)
+ supportedFilters = isPushed.collect { case (true, filter) => filter }.toArray
+ val unsupported = isPushed.collect { case (false, filter) => filter }.toArray
+ unsupported
Review Comment:
what do you mean? supportedFilters ++ unsupported is all filters
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1985500079
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/UserDefinedPythonDataSource.scala:
##########
@@ -300,6 +321,94 @@ private class UserDefinedPythonDataSourceRunner(
}
}
+case class PythonFilterPushdownResult(
+ dataSource: Array[Byte],
+ isFilterPushed: collection.Seq[Boolean])
Review Comment:
`isFilterPushed(i)` means whether filter i is pushed. Maybe I can name it "isFilterPushedMask" to make it clearer.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1990447794
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScan.scala:
##########
@@ -16,26 +16,44 @@
*/
package org.apache.spark.sql.execution.datasources.v2.python
+import org.apache.commons.lang3.StringUtils
+
import org.apache.spark.JobArtifactSet
+import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.connector.metric.CustomMetric
import org.apache.spark.sql.connector.read._
import org.apache.spark.sql.connector.read.streaming.MicroBatchStream
+import org.apache.spark.sql.internal.connector.SupportsMetadata
+import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
-
+import org.apache.spark.util.Utils
class PythonScan(
- ds: PythonDataSourceV2,
- shortName: String,
- outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends Scan {
+ ds: PythonDataSourceV2,
+ shortName: String,
+ outputSchema: StructType,
+ options: CaseInsensitiveStringMap,
+ supportedFilters: Array[Filter]
+) extends Scan
+ with SupportsMetadata {
Review Comment:
```suggestion
) extends Scan with SupportsMetadata {
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1990350885
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScan.scala:
##########
@@ -16,26 +16,43 @@
*/
package org.apache.spark.sql.execution.datasources.v2.python
+import org.apache.commons.lang3.StringUtils
+
import org.apache.spark.JobArtifactSet
+import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.connector.metric.CustomMetric
import org.apache.spark.sql.connector.read._
import org.apache.spark.sql.connector.read.streaming.MicroBatchStream
+import org.apache.spark.sql.internal.connector.SupportsMetadata
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
-
+import org.apache.spark.util.Utils
class PythonScan(
- ds: PythonDataSourceV2,
- shortName: String,
- outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends Scan {
+ ds: PythonDataSourceV2,
+ shortName: String,
+ outputSchema: StructType,
+ options: CaseInsensitiveStringMap,
+ metadata: => Map[String, String]
+) extends Scan
+ with SupportsMetadata {
+ private lazy val sparkSession = SparkSession.active
+ private def maxMetadataValueLength = sparkSession.sessionState.conf.maxMetadataStringLength
Review Comment:
yeah it can be a local val instead
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r2002218436
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonDataSourceSuite.scala:
##########
@@ -213,6 +220,66 @@ class PythonDataSourceSuite extends PythonDataSourceSuiteBase {
parameters = Map("inputSchema" -> "INT", "dataType" -> "\"INT\""))
}
+ test("data source reader with filter pushdown") {
+ assume(shouldTestPandasUDFs)
+ val dataSourceScript =
+ s"""
+ |from pyspark.sql.datasource import (
+ | DataSource,
+ | DataSourceReader,
+ | EqualTo,
+ | InputPartition,
+ |)
+ |
+ |class SimpleDataSourceReader(DataSourceReader):
+ | def partitions(self):
+ | return [InputPartition(i) for i in range(2)]
+ |
+ | def pushFilters(self, filters):
+ | yield filters[filters.index(EqualTo(("id",), 1))]
+ |
+ | def read(self, partition):
+ | yield (0, partition.value)
+ | yield (1, partition.value)
+ | yield (2, partition.value)
+ |
+ |class SimpleDataSource(DataSource):
+ | def schema(self):
+ | return "id int, partition int"
+ |
+ | def reader(self, schema):
+ | return SimpleDataSourceReader()
+ |""".stripMargin
+ val schema = StructType.fromDDL("id INT, partition INT")
+ val dataSource =
+ createUserDefinedPythonDataSource(name = dataSourceName, pythonScript = dataSourceScript)
+ spark.conf.set(SQLConf.PYTHON_FILTER_PUSHDOWN_ENABLED, true)
Review Comment:
> Can we also add another test case with this config disabled?
We do have Python tests with the config disabled. Is that ok?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1984640102
##########
python/pyspark/sql/datasource.py:
##########
@@ -234,6 +249,62 @@ def streamReader(self, schema: StructType) -> "DataSourceStreamReader":
)
+ColumnPath = Tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class Filter(ABC):
+ """
+ The base class for filters used for filter pushdown.
+
+ .. versionadded: 4.1.0
+
+ Notes
+ -----
+ Column references are represented as a tuple of strings. For example:
+
+ +----------------+----------------------+
+ | Column | Representation |
+ +----------------+----------------------+
+ | `col1` | `("col1",)` |
Review Comment:
this is a python syntax to disambiguate between 1-tuple and expression, removing the comma changes meaning
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on PR #49961:
URL: https://github.com/apache/spark/pull/49961#issuecomment-2707172162
@cloud-fan
> Now to push down filters, we need to create the python batch reader earlier, which means one more round of Python worker communication in the optimizer. I'm wondering that once we finish pushdown, shall we do the planning work immediately and keep PythonDataSourceReadInfo in PythonScan?
Good idea to avoid the extra round of worker. I think that would require some refactoring of the plan_read worker so I'll implement that in a new PR since this PR is already very large.
Also when we add column pruning support we should get partitions in column pruning worker rather than filter pushdown worker.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1974189026
##########
python/pyspark/sql/tests/test_python_datasource.py:
##########
@@ -246,6 +248,137 @@ def reader(self, schema) -> "DataSourceReader":
assertDataFrameEqual(df, [Row(x=0, y="0"), Row(x=1, y="1")])
self.assertEqual(df.select(spark_partition_id()).distinct().count(), 2)
+ def test_filter_pushdown(self):
+ class TestDataSourceReader(DataSourceReader):
+ def __init__(self):
+ self.has_filter = False
+
+ def pushdownFilters(self, filters: List[Filter]) -> Iterable[Filter]:
+ assert set(filters) == {
+ EqualTo(("x",), 1),
+ EqualTo(("y",), 2),
+ }, filters
+ self.has_filter = True
+ # pretend we support x = 1 filter but in fact we don't
+ # so we only return y = 2 filter
+ yield filters[filters.index(EqualTo(("y",), 2))]
+
+ def partitions(self):
+ assert self.has_filter
+ return super().partitions()
+
+ def read(self, partition):
+ assert self.has_filter
+ yield [1, 1]
+ yield [1, 2]
+ yield [2, 2]
+
+ class TestDataSource(DataSource):
+ @classmethod
+ def name(cls):
+ return "test"
+
+ def schema(self):
+ return "x int, y int"
+
+ def reader(self, schema) -> "DataSourceReader":
+ return TestDataSourceReader()
+
+ self.spark.dataSource.register(TestDataSource)
+ df = self.spark.read.format("test").load().filter("x = 1 and y = 2")
Review Comment:
We should also include metadata such as PushedFilters in the description of PythonScan, like what FileScan does.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1972886468
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +26,24 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
+ private var supported: Array[Filter] = Array.empty
+
override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+
+ override def pushFilters(filters: Array[Filter]): Array[Filter] = {
+ val dataSource = ds.getOrCreateDataSourceInPython(shortName, options, Some(outputSchema))
+ val result = ds.source.pushdownFiltersInPython(dataSource, outputSchema, filters)
+ ds.setDataSourceInPython(dataSource.copy(dataSource = result.dataSource))
Review Comment:
Added comments explaining why we need to update the data source and when pushFilters is called
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1976389689
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
Review Comment:
Because Python data source is new, we should use SupportsPushDownV2Filters first.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
Review Comment:
Because Python data source is new, we should use `SupportsPushDownV2Filters` first.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r2002140192
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonDataSourceSuite.scala:
##########
@@ -213,6 +220,66 @@ class PythonDataSourceSuite extends PythonDataSourceSuiteBase {
parameters = Map("inputSchema" -> "INT", "dataType" -> "\"INT\""))
}
+ test("data source reader with filter pushdown") {
+ assume(shouldTestPandasUDFs)
+ val dataSourceScript =
+ s"""
+ |from pyspark.sql.datasource import (
+ | DataSource,
+ | DataSourceReader,
+ | EqualTo,
+ | InputPartition,
+ |)
+ |
+ |class SimpleDataSourceReader(DataSourceReader):
+ | def partitions(self):
+ | return [InputPartition(i) for i in range(2)]
+ |
+ | def pushFilters(self, filters):
+ | yield filters[filters.index(EqualTo(("id",), 1))]
+ |
+ | def read(self, partition):
+ | yield (0, partition.value)
+ | yield (1, partition.value)
+ | yield (2, partition.value)
+ |
+ |class SimpleDataSource(DataSource):
+ | def schema(self):
+ | return "id int, partition int"
+ |
+ | def reader(self, schema):
+ | return SimpleDataSourceReader()
+ |""".stripMargin
+ val schema = StructType.fromDDL("id INT, partition INT")
+ val dataSource =
+ createUserDefinedPythonDataSource(name = dataSourceName, pythonScript = dataSourceScript)
+ spark.conf.set(SQLConf.PYTHON_FILTER_PUSHDOWN_ENABLED, true)
Review Comment:
Let's use `withSQLConf`
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonDataSourceSuite.scala:
##########
@@ -213,6 +220,66 @@ class PythonDataSourceSuite extends PythonDataSourceSuiteBase {
parameters = Map("inputSchema" -> "INT", "dataType" -> "\"INT\""))
}
+ test("data source reader with filter pushdown") {
+ assume(shouldTestPandasUDFs)
+ val dataSourceScript =
+ s"""
+ |from pyspark.sql.datasource import (
+ | DataSource,
+ | DataSourceReader,
+ | EqualTo,
+ | InputPartition,
+ |)
+ |
+ |class SimpleDataSourceReader(DataSourceReader):
+ | def partitions(self):
+ | return [InputPartition(i) for i in range(2)]
+ |
+ | def pushFilters(self, filters):
+ | yield filters[filters.index(EqualTo(("id",), 1))]
+ |
+ | def read(self, partition):
+ | yield (0, partition.value)
+ | yield (1, partition.value)
+ | yield (2, partition.value)
+ |
+ |class SimpleDataSource(DataSource):
+ | def schema(self):
+ | return "id int, partition int"
+ |
+ | def reader(self, schema):
+ | return SimpleDataSourceReader()
+ |""".stripMargin
+ val schema = StructType.fromDDL("id INT, partition INT")
+ val dataSource =
+ createUserDefinedPythonDataSource(name = dataSourceName, pythonScript = dataSourceScript)
+ spark.conf.set(SQLConf.PYTHON_FILTER_PUSHDOWN_ENABLED, true)
Review Comment:
Can we also add another test case with this config disabled?
##########
python/pyspark/sql/tests/test_python_datasource.py:
##########
@@ -246,6 +251,161 @@ def reader(self, schema) -> "DataSourceReader":
assertDataFrameEqual(df, [Row(x=0, y="0"), Row(x=1, y="1")])
self.assertEqual(df.select(spark_partition_id()).distinct().count(), 2)
+ def test_filter_pushdown(self):
+ class TestDataSourceReader(DataSourceReader):
+ def __init__(self):
+ self.has_filter = False
+
+ def pushFilters(self, filters: List[Filter]) -> Iterable[Filter]:
+ assert set(filters) == {
+ EqualTo(("x",), 1),
+ EqualTo(("y",), 2),
+ }, filters
+ self.has_filter = True
+ # pretend we support x = 1 filter but in fact we don't
+ # so we only return y = 2 filter
+ yield filters[filters.index(EqualTo(("y",), 2))]
+
+ def partitions(self):
+ assert self.has_filter
+ return super().partitions()
+
+ def read(self, partition):
+ assert self.has_filter
+ yield [1, 1]
+ yield [1, 2]
+ yield [2, 2]
+
+ class TestDataSource(DataSource):
+ @classmethod
+ def name(cls):
+ return "test"
+
+ def schema(self):
+ return "x int, y int"
+
+ def reader(self, schema) -> "DataSourceReader":
+ return TestDataSourceReader()
+
+ self.spark.conf.set("spark.sql.python.filterPushdown.enabled", True)
Review Comment:
let's use `with self.sql_conf`
##########
python/pyspark/sql/tests/test_python_datasource.py:
##########
@@ -246,6 +251,161 @@ def reader(self, schema) -> "DataSourceReader":
assertDataFrameEqual(df, [Row(x=0, y="0"), Row(x=1, y="1")])
self.assertEqual(df.select(spark_partition_id()).distinct().count(), 2)
+ def test_filter_pushdown(self):
+ class TestDataSourceReader(DataSourceReader):
+ def __init__(self):
+ self.has_filter = False
+
+ def pushFilters(self, filters: List[Filter]) -> Iterable[Filter]:
+ assert set(filters) == {
+ EqualTo(("x",), 1),
+ EqualTo(("y",), 2),
+ }, filters
+ self.has_filter = True
+ # pretend we support x = 1 filter but in fact we don't
+ # so we only return y = 2 filter
+ yield filters[filters.index(EqualTo(("y",), 2))]
+
+ def partitions(self):
+ assert self.has_filter
+ return super().partitions()
+
+ def read(self, partition):
+ assert self.has_filter
+ yield [1, 1]
+ yield [1, 2]
+ yield [2, 2]
+
+ class TestDataSource(DataSource):
+ @classmethod
+ def name(cls):
+ return "test"
+
+ def schema(self):
+ return "x int, y int"
+
+ def reader(self, schema) -> "DataSourceReader":
+ return TestDataSourceReader()
+
+ self.spark.conf.set("spark.sql.python.filterPushdown.enabled", True)
+ self.spark.dataSource.register(TestDataSource)
+ df = self.spark.read.format("test").load().filter("x = 1 and y = 2")
+ # only the y = 2 filter is applied post scan
+ assertDataFrameEqual(df, [Row(x=1, y=2), Row(x=2, y=2)])
+
+ def test_extraneous_filter(self):
+ class TestDataSourceReader(DataSourceReader):
+ def pushFilters(self, filters: List[Filter]) -> Iterable[Filter]:
+ yield EqualTo(("x",), 1)
+
+ def partitions(self):
+ assert False
+
+ def read(self, partition):
+ assert False
+
+ class TestDataSource(DataSource):
+ @classmethod
+ def name(cls):
+ return "test"
+
+ def schema(self):
+ return "x int"
+
+ def reader(self, schema) -> "DataSourceReader":
+ return TestDataSourceReader()
+
+ self.spark.conf.set("spark.sql.python.filterPushdown.enabled", True)
+ self.spark.dataSource.register(TestDataSource)
+ with self.assertRaisesRegex(Exception, "DATA_SOURCE_EXTRANEOUS_FILTERS"):
+ self.spark.read.format("test").load().filter("x = 1").show()
+
+ def test_filter_pushdown_error(self):
+ error_str = "dummy error"
+
+ class TestDataSourceReader(DataSourceReader):
+ def pushFilters(self, filters: List[Filter]) -> Iterable[Filter]:
+ raise Exception(error_str)
+
+ def read(self, partition):
+ yield [1]
+
+ class TestDataSource(DataSource):
+ def schema(self):
+ return "x int"
+
+ def reader(self, schema) -> "DataSourceReader":
+ return TestDataSourceReader()
+
+ self.spark.conf.set("spark.sql.python.filterPushdown.enabled", True)
+ self.spark.dataSource.register(TestDataSource)
+ df = self.spark.read.format("TestDataSource").load().filter("x = 1 or x is null")
+ assertDataFrameEqual(df, [Row(x=1)]) # works when not pushing down filters
+ with self.assertRaisesRegex(Exception, error_str):
+ df.filter("x = 1").explain()
+
+ def test_filter_pushdown_disabled(self):
+ class TestDataSourceReader(DataSourceReader):
+ def pushFilters(self, filters: List[Filter]) -> Iterable[Filter]:
+ assert False
+
+ def read(self, partition):
+ assert False
+
+ class TestDataSource(DataSource):
+ def reader(self, schema) -> "DataSourceReader":
+ return TestDataSourceReader()
+
+ self.spark.conf.set("spark.sql.python.filterPushdown.enabled", False)
Review Comment:
ditto
##########
python/pyspark/sql/tests/test_python_datasource.py:
##########
@@ -246,6 +251,161 @@ def reader(self, schema) -> "DataSourceReader":
assertDataFrameEqual(df, [Row(x=0, y="0"), Row(x=1, y="1")])
self.assertEqual(df.select(spark_partition_id()).distinct().count(), 2)
+ def test_filter_pushdown(self):
+ class TestDataSourceReader(DataSourceReader):
+ def __init__(self):
+ self.has_filter = False
+
+ def pushFilters(self, filters: List[Filter]) -> Iterable[Filter]:
+ assert set(filters) == {
+ EqualTo(("x",), 1),
+ EqualTo(("y",), 2),
+ }, filters
+ self.has_filter = True
+ # pretend we support x = 1 filter but in fact we don't
+ # so we only return y = 2 filter
+ yield filters[filters.index(EqualTo(("y",), 2))]
+
+ def partitions(self):
+ assert self.has_filter
+ return super().partitions()
+
+ def read(self, partition):
+ assert self.has_filter
+ yield [1, 1]
+ yield [1, 2]
+ yield [2, 2]
+
+ class TestDataSource(DataSource):
+ @classmethod
+ def name(cls):
+ return "test"
+
+ def schema(self):
+ return "x int, y int"
+
+ def reader(self, schema) -> "DataSourceReader":
+ return TestDataSourceReader()
+
+ self.spark.conf.set("spark.sql.python.filterPushdown.enabled", True)
+ self.spark.dataSource.register(TestDataSource)
+ df = self.spark.read.format("test").load().filter("x = 1 and y = 2")
+ # only the y = 2 filter is applied post scan
+ assertDataFrameEqual(df, [Row(x=1, y=2), Row(x=2, y=2)])
+
+ def test_extraneous_filter(self):
+ class TestDataSourceReader(DataSourceReader):
+ def pushFilters(self, filters: List[Filter]) -> Iterable[Filter]:
+ yield EqualTo(("x",), 1)
+
+ def partitions(self):
+ assert False
+
+ def read(self, partition):
+ assert False
+
+ class TestDataSource(DataSource):
+ @classmethod
+ def name(cls):
+ return "test"
+
+ def schema(self):
+ return "x int"
+
+ def reader(self, schema) -> "DataSourceReader":
+ return TestDataSourceReader()
+
+ self.spark.conf.set("spark.sql.python.filterPushdown.enabled", True)
Review Comment:
ditto
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonDataSourceSuite.scala:
##########
@@ -213,6 +220,66 @@ class PythonDataSourceSuite extends PythonDataSourceSuiteBase {
parameters = Map("inputSchema" -> "INT", "dataType" -> "\"INT\""))
}
+ test("data source reader with filter pushdown") {
+ assume(shouldTestPandasUDFs)
+ val dataSourceScript =
+ s"""
+ |from pyspark.sql.datasource import (
+ | DataSource,
+ | DataSourceReader,
+ | EqualTo,
+ | InputPartition,
+ |)
+ |
+ |class SimpleDataSourceReader(DataSourceReader):
+ | def partitions(self):
+ | return [InputPartition(i) for i in range(2)]
+ |
+ | def pushFilters(self, filters):
+ | yield filters[filters.index(EqualTo(("id",), 1))]
+ |
+ | def read(self, partition):
+ | yield (0, partition.value)
+ | yield (1, partition.value)
+ | yield (2, partition.value)
+ |
+ |class SimpleDataSource(DataSource):
+ | def schema(self):
+ | return "id int, partition int"
+ |
+ | def reader(self, schema):
+ | return SimpleDataSourceReader()
+ |""".stripMargin
+ val schema = StructType.fromDDL("id INT, partition INT")
+ val dataSource =
+ createUserDefinedPythonDataSource(name = dataSourceName, pythonScript = dataSourceScript)
+ spark.conf.set(SQLConf.PYTHON_FILTER_PUSHDOWN_ENABLED, true)
+ spark.dataSource.registerPython(dataSourceName, dataSource)
+ val df =
+ spark.read.format(dataSourceName).schema(schema).load().filter("id = 1 and partition = 0")
+ val plan = df.queryExecution.executedPlan
+
+ val filter = collectFirst(df.queryExecution.executedPlan) {
Review Comment:
Please also add the executed plan in the comment here.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1985908707
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
+ private var supportedFilters: Array[Filter] = Array.empty
+
+ private def metadata: Map[String, String] = {
+ Map(
+ "PushedFilters" -> supportedFilters.mkString("[", ", ", "]"),
+ "ReadSchema" -> outputSchema.simpleString
+ )
+ }
+
+ override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options, metadata)
Review Comment:
Shall we add `supportedFilters` into the constructor of `PythonScan` and remove metadata from `PythonScanBuilder`.
We can override `getMetaData` in `PythonScan`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1985910327
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/UserDefinedPythonDataSource.scala:
##########
@@ -300,6 +321,94 @@ private class UserDefinedPythonDataSourceRunner(
}
}
+case class PythonFilterPushdownResult(
+ dataSource: Array[Byte],
+ isFilterPushed: collection.Seq[Boolean])
Review Comment:
How about `filterPushedStatuses` ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1984617330
##########
python/pyspark/sql/datasource.py:
##########
@@ -234,6 +249,62 @@ def streamReader(self, schema: StructType) -> "DataSourceStreamReader":
)
+ColumnPath = Tuple[str, ...]
+
+
+@dataclass(frozen=True)
Review Comment:
what does `frozen=True` mean?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1984636973
##########
python/pyspark/sql/datasource.py:
##########
@@ -234,6 +249,62 @@ def streamReader(self, schema: StructType) -> "DataSourceStreamReader":
)
+ColumnPath = Tuple[str, ...]
+
+
+@dataclass(frozen=True)
Review Comment:
means that it's immutable (we don't want users to think that they can mutate pushed filters)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1984797430
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
+ private var supportedFilters: Array[Filter] = Array.empty
+
+ private def metadata: Map[String, String] = {
+ Map(
+ "PushedFilters" -> supportedFilters.mkString("[", ", ", "]"),
+ "ReadSchema" -> outputSchema.simpleString
+ )
+ }
+
+ override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options, metadata)
+
+ // Optionally called by DSv2 once to push down filters before the scan is built.
+ override def pushFilters(filters: Array[Filter]): Array[Filter] = {
+ if (!SQLConf.get.pythonFilterPushDown) {
+ return filters
+ }
+
+ val dataSource = ds.getOrCreateDataSourceInPython(shortName, options, Some(outputSchema))
+ val result = ds.source.pushdownFiltersInPython(dataSource, outputSchema, filters)
+
+ // The Data Source instance state changes after pushdown to remember the reader instance
+ // created and the filters pushed down. So pushdownFiltersInPython returns a new pickled
+ // Data Source instance. We need to use that new instance for further operations.
+ ds.setDataSourceInPython(dataSource.copy(dataSource = result.dataSource))
+
+ // Partition the filters into supported and unsupported ones.
+ val isPushed = result.isFilterPushed.zip(filters)
+ supportedFilters = isPushed.collect { case (true, filter) => filter }.toArray
+ val unsupported = isPushed.collect { case (false, filter) => filter }.toArray
+ unsupported
Review Comment:
Do we need keep all the filters, it will be more robust.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
+ private var supportedFilters: Array[Filter] = Array.empty
+
+ private def metadata: Map[String, String] = {
+ Map(
+ "PushedFilters" -> supportedFilters.mkString("[", ", ", "]"),
+ "ReadSchema" -> outputSchema.simpleString
+ )
+ }
+
+ override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options, metadata)
+
+ // Optionally called by DSv2 once to push down filters before the scan is built.
+ override def pushFilters(filters: Array[Filter]): Array[Filter] = {
+ if (!SQLConf.get.pythonFilterPushDown) {
+ return filters
+ }
+
+ val dataSource = ds.getOrCreateDataSourceInPython(shortName, options, Some(outputSchema))
+ val result = ds.source.pushdownFiltersInPython(dataSource, outputSchema, filters)
+
+ // The Data Source instance state changes after pushdown to remember the reader instance
+ // created and the filters pushed down. So pushdownFiltersInPython returns a new pickled
+ // Data Source instance. We need to use that new instance for further operations.
+ ds.setDataSourceInPython(dataSource.copy(dataSource = result.dataSource))
+
+ // Partition the filters into supported and unsupported ones.
+ val isPushed = result.isFilterPushed.zip(filters)
+ supportedFilters = isPushed.collect { case (true, filter) => filter }.toArray
+ val unsupported = isPushed.collect { case (false, filter) => filter }.toArray
+ unsupported
Review Comment:
Do we need keep all the filters? it will be more robust.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1985529476
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScanBuilder.scala:
##########
@@ -25,6 +27,40 @@ class PythonScanBuilder(
ds: PythonDataSourceV2,
shortName: String,
outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends ScanBuilder {
- override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options)
+ options: CaseInsensitiveStringMap)
+ extends ScanBuilder
+ with SupportsPushDownFilters {
+ private var supportedFilters: Array[Filter] = Array.empty
+
+ private def metadata: Map[String, String] = {
+ Map(
+ "PushedFilters" -> supportedFilters.mkString("[", ", ", "]"),
+ "ReadSchema" -> outputSchema.simpleString
+ )
+ }
+
+ override def build(): Scan = new PythonScan(ds, shortName, outputSchema, options, metadata)
+
+ // Optionally called by DSv2 once to push down filters before the scan is built.
+ override def pushFilters(filters: Array[Filter]): Array[Filter] = {
+ if (!SQLConf.get.pythonFilterPushDown) {
Review Comment:
We'd like to avoid the new code path (serializing filters, running new python worker, ...) for existing Python data sources that don't implement pushdown. So in case there's a crash or a performance issue in the new code path, its impact is limited.
However we currently don't have a good way to detect whether user has implemented `pushFilters()` in Python `DataSourceReader` before `ScanBuilder.pushFilters()` is called. This is because we don't know whether it's a streaming read or a batch read at this point (the optimizer knows but the data source doesn't get this info) so it's not safe to call Python `DataSource.reader()` to get the batch reader instance.
So we instead add a conf to turn off the new code path. But also if the user imeplements `pushFilters()` and this conf is disabled then we throw an error to let the user know that they must turn on the conf to enable filter pushdown.
In the future if we figure out how to check whether the Python reader imeplements `pushFilters` then we can set this conf to enabled by default and deprecate it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r1972838640
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -69,7 +70,7 @@ def latest_offset_func(reader: DataSourceStreamReader, outfile: IO) -> None:
def partitions_func(
reader: DataSourceStreamReader,
- data_source: DataSource,
+ data_source_name: str,
Review Comment:
I reverted that change
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r2002744212
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/python/PythonScan.scala:
##########
@@ -16,26 +16,43 @@
*/
package org.apache.spark.sql.execution.datasources.v2.python
+import org.apache.commons.lang3.StringUtils
+
import org.apache.spark.JobArtifactSet
+import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.connector.metric.CustomMetric
import org.apache.spark.sql.connector.read._
import org.apache.spark.sql.connector.read.streaming.MicroBatchStream
+import org.apache.spark.sql.internal.connector.SupportsMetadata
+import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
-
+import org.apache.spark.util.Utils
class PythonScan(
- ds: PythonDataSourceV2,
- shortName: String,
- outputSchema: StructType,
- options: CaseInsensitiveStringMap) extends Scan {
+ ds: PythonDataSourceV2,
+ shortName: String,
+ outputSchema: StructType,
+ options: CaseInsensitiveStringMap,
+ supportedFilters: Array[Filter]
+) extends Scan with SupportsMetadata {
+ private lazy val sparkSession = SparkSession.active
override def toBatch: Batch = new PythonBatch(ds, shortName, outputSchema, options)
override def toMicroBatchStream(checkpointLocation: String): MicroBatchStream =
new PythonMicroBatchStream(ds, shortName, outputSchema, options)
- override def description: String = "(Python)"
+ override def description: String = {
Review Comment:
+1
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r2003793967
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonDataSourceSuite.scala:
##########
@@ -213,6 +220,66 @@ class PythonDataSourceSuite extends PythonDataSourceSuiteBase {
parameters = Map("inputSchema" -> "INT", "dataType" -> "\"INT\""))
}
+ test("data source reader with filter pushdown") {
+ assume(shouldTestPandasUDFs)
+ val dataSourceScript =
+ s"""
+ |from pyspark.sql.datasource import (
+ | DataSource,
+ | DataSourceReader,
+ | EqualTo,
+ | InputPartition,
+ |)
+ |
+ |class SimpleDataSourceReader(DataSourceReader):
+ | def partitions(self):
+ | return [InputPartition(i) for i in range(2)]
+ |
+ | def pushFilters(self, filters):
+ | yield filters[filters.index(EqualTo(("id",), 1))]
+ |
+ | def read(self, partition):
+ | yield (0, partition.value)
+ | yield (1, partition.value)
+ | yield (2, partition.value)
+ |
+ |class SimpleDataSource(DataSource):
+ | def schema(self):
+ | return "id int, partition int"
+ |
+ | def reader(self, schema):
+ | return SimpleDataSourceReader()
+ |""".stripMargin
+ val schema = StructType.fromDDL("id INT, partition INT")
+ val dataSource =
+ createUserDefinedPythonDataSource(name = dataSourceName, pythonScript = dataSourceScript)
+ spark.conf.set(SQLConf.PYTHON_FILTER_PUSHDOWN_ENABLED, true)
+ spark.dataSource.registerPython(dataSourceName, dataSource)
+ val df =
+ spark.read.format(dataSourceName).schema(schema).load().filter("id = 1 and partition = 0")
+ val plan = df.queryExecution.executedPlan
+
+ val filter = collectFirst(df.queryExecution.executedPlan) {
Review Comment:
```
== Physical Plan ==
*(1) Project [id#261, partition#262]
+- *(1) Filter ((isnotnull(id#261) AND isnotnull(partition#262)) AND (id#261 = 1))
+- BatchScan SimpleDataSource[id#261, partition#262] (Python)
PushedFilters: [EqualTo(partition,0)],
ReadSchema: struct<id:int,partition:int> RuntimeFilters: []
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51271][PYTHON] Add filter pushdown API to Python Data Sources [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49961:
URL: https://github.com/apache/spark/pull/49961#discussion_r2003792385
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -4673,6 +4673,13 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val PYTHON_FILTER_PUSHDOWN_ENABLED = buildConf("spark.sql.python.filterPushdown.enabled")
+ .doc("When true, enable filter pushdown to Python datasource, at the cost of running " +
+ "Python worker one additional time during planning.")
Review Comment:
Currently this is still true. I have a separate PR to combine filter pushdown & plan read.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org