[SNOW-1791241] Add a session parameter to disable scoped read only te…

…mp table (#2587)

<!---
Please answer these questions before creating your pull request. Thanks!
--->

1. Which Jira issue is this PR addressing? Make sure that there is an
accompanying issue to your PR.

   <!---
   In this section, please add a Snowflake Jira issue number.

Note that if a corresponding GitHub issue exists, you should still
include
   the Snowflake Jira issue number. For example, for GitHub issue
#1400, you should
   add "SNOW-1335071" here.
    --->

SNOW-1791241

2. Fill out the following pre-review checklist:

- [ ] I am adding a new automated test(s) to verify correctness of my
new code
- [ ] If this test skips Local Testing mode, I'm requesting review from
@snowflakedb/local-testing
   - [ ] I am adding new logging messages
   - [ ] I am adding a new telemetry message
   - [ ] I am adding new credentials
   - [ ] I am adding a new dependency
- [ ] If this is a new feature/behavior, I'm adding the Local Testing
parity changes.
- [x] I acknowledge that I have ensured my changes to be thread-safe.
Follow the link for more information: [Thread-safe Developer
Guidelines](https://docs.google.com/document/d/162d_i4zZ2AfcGRXojj0jByt8EUq-DrSHPPnTa4QvwbA/edit#bookmark=id.e82u4nekq80k)

3. Please describe how your code solves the related issue.
scoped read only table doesn't work with dynamic pivot in notebook
today, investigation needed.
We added a flag to turns the scoped temp read only table off for
snowpark padnas.

Follow up: https://snowflakecomputing.atlassian.net/browse/SNOW-1793002
enable the flag at server side for native app

CHANGELOG.md

@@ -54,6 +54,7 @@
   - Treat an empty `subset` (e.g. `[]`) as if it specified all columns instead of no columns.
   - Raise a `TypeError` for a scalar `subset` instead of filtering on just that column.
   - Raise a `ValueError` for a `subset` of type `pandas.Index` instead of filtering on the columns in the index.
+- Disable creation of scoped read only table to mitigate Disable creation of scoped read only table to mitigate `TableNotFoundError` when using dynamic pivot in notebook environment.
 
 #### Improvements
 - Improve np.where with scalar x value by eliminating unnecessary join and temp table creation.

src/snowflake/snowpark/modin/plugin/_internal/utils.py

@@ -273,7 +273,7 @@ def _create_read_only_table(
     readonly_table_name = (
         f"{random_name_for_temp_object(TempObjectType.TABLE)}{READ_ONLY_TABLE_SUFFIX}"
     )
-    use_scoped_temp_table = session._use_scoped_temp_objects
+    use_scoped_temp_table = session._use_scoped_temp_read_only_table
     # If we need to materialize into a temp table our create table expression
     # needs to be SELECT * FROM (object).
     if materialize_into_temp_table:

src/snowflake/snowpark/session.py

@@ -238,6 +238,10 @@
 _PYTHON_SNOWPARK_ENABLE_THREAD_SAFE_SESSION = (
     "PYTHON_SNOWPARK_ENABLE_THREAD_SAFE_SESSION"
 )
+# Flag for controlling the usage of scoped temp read only table.
+_PYTHON_SNOWPARK_ENABLE_SCOPED_TEMP_READ_ONLY_TABLE = (
+    "PYTHON_SNOWPARK_ENABLE_SCOPED_TEMP_READ_ONLY_TABLE"
+)
 # The complexity score lower bound is set to match COMPILATION_MEMORY_LIMIT
 # in Snowflake. This is the limit where we start seeing compilation errors.
 DEFAULT_COMPLEXITY_SCORE_LOWER_BOUND = 10_000_000
@@ -541,6 +545,11 @@ def __init__(
                 _PYTHON_SNOWPARK_USE_SCOPED_TEMP_OBJECTS_STRING, True
             )
         )
+        self._use_scoped_temp_read_only_table: bool = (
+            self._conn._get_client_side_session_parameter(
+                _PYTHON_SNOWPARK_ENABLE_SCOPED_TEMP_READ_ONLY_TABLE, False
+            )
+        )
         self._file = FileOperation(self)
         self._lineage = Lineage(self)
         self._sql_simplifier_enabled: bool = (

tests/integ/modin/io/test_read_snowflake.py

@@ -39,10 +39,10 @@
 
 @pytest.fixture(params=paramList)
 def setup_use_scoped_object(request, session):
-    use_scoped_objects = session._use_scoped_temp_objects
-    session._use_scoped_temp_objects = request.param
+    use_scoped_objects = session._use_scoped_temp_read_only_table
+    session._use_scoped_temp_read_only_table = request.param
     yield
-    session._use_scoped_temp_objects = use_scoped_objects
+    session._use_scoped_temp_read_only_table = use_scoped_objects
 
 
 def read_snowflake_and_verify_snapshot_creation(
@@ -85,7 +85,7 @@ def read_snowflake_and_verify_snapshot_creation(
         assert len(query_history.queries) == 1
 
     # test if the scoped snapshot is created
-    scoped_pattern = " SCOPED " if session._use_scoped_temp_objects else " "
+    scoped_pattern = " SCOPED " if session._use_scoped_temp_read_only_table else " "
     table_create_sql = query_history.queries[-1].sql_text
     table_create_pattern = f"CREATE OR REPLACE{scoped_pattern}TEMPORARY READ ONLY TABLE SNOWPARK_TEMP_TABLE_[0-9A-Z]+.*{READ_ONLY_TABLE_SUFFIX}.*"
     assert re.match(table_create_pattern, table_create_sql) is not None