Skip to content
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions python/pyspark/sql/tests/arrow/test_arrow_udf_scalar.py
Original file line number Diff line number Diff line change
Expand Up @@ -1267,6 +1267,33 @@ def return_one(iterator):
result = df.select(return_one("id").alias("one")).collect()
self.assertEqual(expected, result)

def test_arrow_udf_task_context(self):
import pyarrow as pa

from pyspark import TaskContext

def partition_ids(size):
task_context = TaskContext.get()
assert task_context is not None
return pa.array([task_context.partitionId()] * size, type=pa.int32())

@arrow_udf("int")
def scalar_task_context(values):
return partition_ids(len(values))

@arrow_udf("int", ArrowUDFType.SCALAR_ITER)
def scalar_iter_task_context(iterator):
for values in iterator:
yield partition_ids(len(values))

for task_context_udf in [scalar_task_context, scalar_iter_task_context]:
rows = (
self.spark.range(10, numPartitions=2)
.select(task_context_udf("id").alias("partition_id"))
.collect()
)
self.assertEqual({0, 1}, {row.partition_id for row in rows})


class ScalarArrowUDFTests(ScalarArrowUDFTestsMixin, ReusedSQLTestCase):
@classmethod
Expand Down