From 977676137dddba16c07fa1a9398dacb0ca056db2 Mon Sep 17 00:00:00 2001 From: MBM Date: Thu, 21 May 2026 16:27:29 +0530 Subject: [PATCH 1/4] fix: fix signal.signal() ValueError in Trino worker threads Signed-off-by: MBM --- .../tests/test_trino_queries.py | 37 +++++++++++++++++++ .../trino_offline_store/trino_queries.py | 6 ++- 2 files changed, 41 insertions(+), 2 deletions(-) create mode 100644 sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py diff --git a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py new file mode 100644 index 00000000000..883ce71ada1 --- /dev/null +++ b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py @@ -0,0 +1,37 @@ +import threading +from unittest.mock import MagicMock + +from feast.infra.offline_stores.contrib.trino_offline_store.trino_queries import ( + Query, +) + + +def test_query_init_in_main_thread_registers_signals(): + """signal.signal() should work fine in main thread.""" + cursor = MagicMock() + # Should not raise any exception in main thread + query = Query(query_text="SELECT 1", cursor=cursor) + assert query.query_text == "SELECT 1" + + +def test_query_init_in_worker_thread_does_not_raise(): + """Regression test: signal.signal() fails in non-main threads.""" + # signal.signal() raises ValueError when called outside the main thread. + # This test verifies the fix guards against that by running Query.__init__ + # in a worker thread and ensuring no exception is raised. + + errors = [] + cursor = MagicMock() + + def create_query(): + try: + query = Query(query_text="SELECT 1", cursor=cursor) + assert query.query_text == "SELECT 1" + except ValueError as e: + errors.append(e) + + thread = threading.Thread(target=create_query) + thread.start() + thread.join() + + assert not errors, f"Unexpected ValueError in worker thread: {errors[0]}" diff --git a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino_queries.py b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino_queries.py index 302745fc0e9..ef23a5cf021 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino_queries.py +++ b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino_queries.py @@ -1,6 +1,7 @@ from __future__ import annotations import signal +import threading from dataclasses import dataclass from enum import Enum from typing import Any, Dict, List, Optional @@ -92,8 +93,9 @@ def __init__(self, query_text: str, cursor: Cursor): self.status = QueryStatus.PENDING self._cursor = cursor - signal.signal(signal.SIGINT, self.cancel) - signal.signal(signal.SIGTERM, self.cancel) + if threading.current_thread() is threading.main_thread(): + signal.signal(signal.SIGINT, self.cancel) + signal.signal(signal.SIGTERM, self.cancel) def execute(self) -> Results: try: From c5ca3f8fc3423f1a6d589be355bbbfe6408d0640 Mon Sep 17 00:00:00 2001 From: MBM Date: Mon, 1 Jun 2026 15:01:56 +0530 Subject: [PATCH 2/4] test: patch signal.signal in main thread test to avoid handler leak Signed-off-by: MBM --- .../trino_offline_store/tests/test_trino_queries.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py index 883ce71ada1..ee2284ea55f 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py +++ b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py @@ -1,5 +1,5 @@ import threading -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch from feast.infra.offline_stores.contrib.trino_offline_store.trino_queries import ( Query, @@ -8,10 +8,14 @@ def test_query_init_in_main_thread_registers_signals(): """signal.signal() should work fine in main thread.""" - cursor = MagicMock() + # Should not raise any exception in main thread - query = Query(query_text="SELECT 1", cursor=cursor) - assert query.query_text == "SELECT 1" + cursor = MagicMock() + with patch("signal.signal") as mock_signal: + query = Query(query_text="SELECT 1", cursor=cursor) + assert query.query_text == "SELECT 1" + # Expected signal.signal to be called twice for SIGINT and SIGTERM + assert mock_signal.call_count == 2 def test_query_init_in_worker_thread_does_not_raise(): From d9d3faba4baa3c76970db781a8946f304187d5cf Mon Sep 17 00:00:00 2001 From: mbm-codes <59091429+mbm-codes@users.noreply.github.com> Date: Wed, 3 Jun 2026 13:23:22 +0530 Subject: [PATCH 3/4] Update sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py Apply reviewer suggestion: verify correct signals and handler in test Co-authored-by: Jitendra Yejare Signed-off-by: MBM --- .../trino_offline_store/tests/test_trino_queries.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py index ee2284ea55f..e8c9ba71c01 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py +++ b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py @@ -12,9 +12,10 @@ def test_query_init_in_main_thread_registers_signals(): # Should not raise any exception in main thread cursor = MagicMock() with patch("signal.signal") as mock_signal: - query = Query(query_text="SELECT 1", cursor=cursor) - assert query.query_text == "SELECT 1" - # Expected signal.signal to be called twice for SIGINT and SIGTERM + # Verify signal handlers are registered correctly + assert mock_signal.call_count == 2 + mock_signal.assert_any_call(signal.SIGINT, query.cancel) + mock_signal.assert_any_call(signal.SIGTERM, query.cancel) assert mock_signal.call_count == 2 From 4c1e5d8c0d036746ec45e9074539f860e87dc1a9 Mon Sep 17 00:00:00 2001 From: MBM Date: Wed, 3 Jun 2026 14:25:03 +0530 Subject: [PATCH 4/4] test: address PR feedback for worker thread errors and signals - Simplify worker thread assertion message to safely print errors list - Add verification for signal handler registration details Signed-off-by: MBM --- .../trino_offline_store/tests/test_trino_queries.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py index e8c9ba71c01..2f7071b6bd6 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py +++ b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_queries.py @@ -1,3 +1,4 @@ +import signal import threading from unittest.mock import MagicMock, patch @@ -11,11 +12,16 @@ def test_query_init_in_main_thread_registers_signals(): # Should not raise any exception in main thread cursor = MagicMock() + with patch("signal.signal") as mock_signal: + query = Query(query_text="SELECT 1", cursor=cursor) + assert query.query_text == "SELECT 1" + # Verify signal handlers are registered correctly - assert mock_signal.call_count == 2 mock_signal.assert_any_call(signal.SIGINT, query.cancel) mock_signal.assert_any_call(signal.SIGTERM, query.cancel) + + # Expected signal.signal to be called twice for SIGINT and SIGTERM assert mock_signal.call_count == 2 @@ -39,4 +45,4 @@ def create_query(): thread.start() thread.join() - assert not errors, f"Unexpected ValueError in worker thread: {errors[0]}" + assert not errors, f"Unexpected ValueError in worker thread: {errors}"