Added flag to allow override SnowflakeDialect div_is_floor_div flag

DESCRIPTION.md

@@ -11,8 +11,12 @@ Source code is also available at:
 
 - (Unreleased)
   - Add support for partition by to copy into <location>
+  - Added `force_div_is_floordiv` flag to override `div_is_floordiv` new default value `False` in `SnowflakeDialect`.
+    - With the flag in `False`, the `/` division operator will be treated as a float division and `//` as a floor division.
+    - This flag is added to maintain backward compatibility with the previous behavior of Snowflake Dialect division.
+    - This flag will be removed in the future and Snowflake Dialect will use `div_is_floor_div` as `False`.
 
-- v1.7.0(November 22, 2024)
+- v1.7.0(November 21, 2024)
 
   - Add support for dynamic tables and required options
   - Add support for hybrid tables

pyproject.toml

@@ -84,6 +84,11 @@ extra-dependencies = ["SQLAlchemy>=1.4.19,<2.0.0"]
 features = ["development", "pandas"]
 python = "3.8"
 
+[tool.hatch.envs.sa14.scripts]
+test-dialect = "pytest --ignore_v20_test -ra -vvv --tb=short --cov snowflake.sqlalchemy --cov-append --junitxml ./junit.xml --ignore=tests/sqlalchemy_test_suite tests/"
+test-dialect-compatibility = "pytest --ignore_v20_test -ra -vvv --tb=short --cov snowflake.sqlalchemy --cov-append --junitxml ./junit.xml tests/sqlalchemy_test_suite"
+test-dialect-aws = "pytest --ignore_v20_test -m \"aws\" -ra -vvv --tb=short --cov snowflake.sqlalchemy --cov-append --junitxml ./junit.xml --ignore=tests/sqlalchemy_test_suite tests/"
+
 [tool.hatch.envs.default.env-vars]
 COVERAGE_FILE = "coverage.xml"
 SQLACHEMY_WARN_20 = "1"
@@ -131,4 +136,5 @@ markers = [
   "requires_external_volume: tests that needs a external volume to be executed",
   "external: tests that could but should only run on our external CI",
   "feature_max_lob_size: tests that could but should only run on our external CI",
+  "feature_v20: tests that could but should only run on SqlAlchemy v20",
 ]

src/snowflake/sqlalchemy/base.py

@@ -5,6 +5,7 @@
 import itertools
 import operator
 import re
+import warnings
 from typing import List
 
 from sqlalchemy import exc as sa_exc
@@ -799,6 +800,15 @@ def visit_join(self, join, asfrom=False, from_linter=None, **kwargs):
             + join.onclause._compiler_dispatch(self, from_linter=from_linter, **kwargs)
         )
 
+    def visit_floordiv_binary(self, binary, operator, **kw):
+        if self.dialect.div_is_floordiv and IS_VERSION_20:
+            warnings.warn(
+                "div_is_floordiv value will be changed to False in a future release. This will generate a behavior change on true and floor division. Please review https://docs.sqlalchemy.org/en/20/changelog/whatsnew_20.html#python-division-operator-performs-true-division-for-all-backends-added-floor-division",
+                PendingDeprecationWarning,
+                stacklevel=2,
+            )
+        return super().visit_floordiv_binary(binary, operator, **kw)
+
     def render_literal_value(self, value, type_):
         # escape backslash
         return super().render_literal_value(value, type_).replace("\\", "\\\\")

src/snowflake/sqlalchemy/snowdialect.py

@@ -71,6 +71,9 @@ class SnowflakeDialect(default.DefaultDialect):
     colspecs = colspecs
     ischema_names = ischema_names
 
+    # target database treats the / division operator as “floor division”
+    div_is_floordiv = False
+
     # all str types must be converted in Unicode
     convert_unicode = True
 
@@ -137,6 +140,19 @@ class SnowflakeDialect(default.DefaultDialect):
 
     supports_identity_columns = True
 
+    def __init__(
+        self,
+        force_div_is_floordiv: bool = True,
+        **kwargs,
+    ):
+        default.DefaultDialect.__init__(self, **kwargs)
+        self.force_div_is_floordiv = force_div_is_floordiv
+        self.div_is_floordiv = force_div_is_floordiv
+
+    def initialize(self, connection):
+        super().initialize(connection)
+        self.div_is_floordiv = self.force_div_is_floordiv
+
     @classmethod
     def dbapi(cls):
         return cls.import_dbapi()

tests/conftest.py

@@ -46,6 +46,26 @@
 TEST_SCHEMA = f"sqlalchemy_tests_{str(uuid.uuid4()).replace('-', '_')}"
 
 
+def pytest_addoption(parser):
+    parser.addoption(
+        "--ignore_v20_test",
+        action="store_true",
+        default=False,
+        help="skip sqlalchemy 2.0 exclusive tests",
+    )
+
+
+def pytest_collection_modifyitems(config, items):
+    if config.getoption("--ignore_v20_test"):
+        # --ignore_v20_test given in cli: skip sqlalchemy 2.0 tests
+        skip_feature_v2 = pytest.mark.skip(
+            reason="need remove --ignore_v20_test option to run"
+        )
+        for item in items:
+            if "feature_v20" in item.keywords:
+                item.add_marker(skip_feature_v2)
+
+
 @pytest.fixture(scope="session")
 def on_travis():
     return os.getenv("TRAVIS", "").lower() == "true"

tests/test_compiler.py

@@ -2,12 +2,14 @@
 # Copyright (c) 2012-2023 Snowflake Computing Inc. All rights reserved.
 #
 
+import pytest
 from sqlalchemy import Integer, String, and_, func, select
 from sqlalchemy.schema import DropColumnComment, DropTableComment
 from sqlalchemy.sql import column, quoted_name, table
 from sqlalchemy.testing.assertions import AssertsCompiledSQL
 
 from snowflake.sqlalchemy import snowdialect
+from src.snowflake.sqlalchemy.snowdialect import SnowflakeDialect
 
 table1 = table(
     "table1", column("id", Integer), column("name", String), column("value", Integer)
@@ -120,3 +122,85 @@ def test_outer_lateral_join():
         str(stmt.compile(dialect=snowdialect.dialect()))
         == "SELECT colname AS label \nFROM abc JOIN LATERAL flatten(PARSE_JSON(colname2)) AS anon_1 GROUP BY colname"
     )
+
+
+@pytest.mark.feature_v20
+def test_division_operator_with_force_div_is_floordiv_false():
+    col1 = column("col1", Integer)
+    col2 = column("col2", Integer)
+    stmt = col1 / col2
+    assert (
+        str(stmt.compile(dialect=SnowflakeDialect(force_div_is_floordiv=False)))
+        == "col1 / col2"
+    )
+
+
+@pytest.mark.feature_v20
+def test_division_operator_with_denominator_expr_force_div_is_floordiv_false():
+    col1 = column("col1", Integer)
+    col2 = column("col2", Integer)
+    stmt = col1 / func.sqrt(col2)
+    assert (
+        str(stmt.compile(dialect=SnowflakeDialect(force_div_is_floordiv=False)))
+        == "col1 / sqrt(col2)"
+    )
+
+
+@pytest.mark.feature_v20
+def test_division_operator_with_force_div_is_floordiv_default_true():
+    col1 = column("col1", Integer)
+    col2 = column("col2", Integer)
+    stmt = col1 / col2
+    assert (
+        str(stmt.compile(dialect=SnowflakeDialect())) == "col1 / CAST(col2 AS NUMERIC)"
+    )
+
+
+@pytest.mark.feature_v20
+def test_division_operator_with_denominator_expr_force_div_is_floordiv_default_true():
+    col1 = column("col1", Integer)
+    col2 = column("col2", Integer)
+    stmt = col1 / func.sqrt(col2)
+    assert (
+        str(stmt.compile(dialect=SnowflakeDialect()))
+        == "col1 / CAST(sqrt(col2) AS NUMERIC)"
+    )
+
+
+@pytest.mark.feature_v20
+def test_floor_division_operator_force_div_is_floordiv_false():
+    col1 = column("col1", Integer)
+    col2 = column("col2", Integer)
+    stmt = col1 // col2
+    assert (
+        str(stmt.compile(dialect=SnowflakeDialect(force_div_is_floordiv=False)))
+        == "FLOOR(col1 / col2)"
+    )
+
+
+@pytest.mark.feature_v20
+def test_floor_division_operator_with_denominator_expr_force_div_is_floordiv_false():
+    col1 = column("col1", Integer)
+    col2 = column("col2", Integer)
+    stmt = col1 // func.sqrt(col2)
+    assert (
+        str(stmt.compile(dialect=SnowflakeDialect(force_div_is_floordiv=False)))
+        == "FLOOR(col1 / sqrt(col2))"
+    )
+
+
+@pytest.mark.feature_v20
+def test_floor_division_operator_force_div_is_floordiv_default_true():
+    col1 = column("col1", Integer)
+    col2 = column("col2", Integer)
+    stmt = col1 // col2
+    assert str(stmt.compile(dialect=SnowflakeDialect())) == "col1 / col2"
+
+
+@pytest.mark.feature_v20
+def test_floor_division_operator_with_denominator_expr_force_div_is_floordiv_default_true():
+    col1 = column("col1", Integer)
+    col2 = column("col2", Integer)
+    stmt = col1 // func.sqrt(col2)
+    res = stmt.compile(dialect=SnowflakeDialect())
+    assert str(res) == "FLOOR(col1 / sqrt(col2))"

tests/test_core.py

@@ -36,8 +36,9 @@
     text,
 )
 from sqlalchemy.exc import DBAPIError, NoSuchTableError, OperationalError
-from sqlalchemy.sql import and_, not_, or_, select
+from sqlalchemy.sql import and_, literal, not_, or_, select
 from sqlalchemy.sql.ddl import CreateTable
+from sqlalchemy.testing.assertions import eq_
 
 import snowflake.connector.errors
 import snowflake.sqlalchemy.snowdialect
@@ -1863,3 +1864,31 @@ def test_snowflake_sqlalchemy_as_valid_client_type():
         snowflake.connector.connection.DEFAULT_CONFIGURATION[
             "internal_application_version"
         ] = origin_internal_app_version
+
+
+@pytest.mark.feature_v20
+def test_division_force_div_is_floordiv_default():
+    engine = create_engine(URL(**CONNECTION_PARAMETERS))
+    expected_warning = "div_is_floordiv value will be changed to False in a future release. This will generate a behavior change on true and floor division. Please review https://docs.sqlalchemy.org/en/20/changelog/whatsnew_20.html#python-division-operator-performs-true-division-for-all-backends-added-floor-division"
+    with pytest.warns(PendingDeprecationWarning, match=expected_warning):
+        with engine.connect() as conn:
+            eq_(
+                conn.execute(
+                    select(literal(5) / literal(10), literal(5) // literal(10))
+                ).fetchall(),
+                [(0.5, 0.5)],
+            )
+
+
+@pytest.mark.feature_v20
+def test_division_force_div_is_floordiv_false():
+    engine = create_engine(
+        URL(**CONNECTION_PARAMETERS), **{"force_div_is_floordiv": False}
+    )
+    with engine.connect() as conn:
+        eq_(
+            conn.execute(
+                select(literal(5) / literal(10), literal(5) // literal(10))
+            ).fetchall(),
+            [(0.5, 0)],
+        )