Add tests to try max lob size in memory feature (#529)

* Add test with large object

pyproject.toml

@@ -109,6 +109,7 @@ line-length = 88
 line-length = 88
 
 [tool.pytest.ini_options]
+addopts = "-m 'not feature_max_lob_size'"
 markers = [
   # Optional dependency groups markers
   "lambda: AWS lambda tests",
@@ -126,4 +127,5 @@ markers = [
   "timeout: tests that need a timeout time",
   "internal: tests that could but should only run on our internal CI",
   "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",
 ]

tests/test_custom_types.py

@@ -2,7 +2,10 @@
 # Copyright (c) 2012-2023 Snowflake Computing Inc. All rights reserved.
 #
 
-from snowflake.sqlalchemy import custom_types
+import pytest
+from sqlalchemy import Column, Integer, MetaData, Table, text
+
+from snowflake.sqlalchemy import TEXT, custom_types
 
 
 def test_string_conversions():
@@ -34,3 +37,31 @@ def test_string_conversions():
         sample = getattr(custom_types, type_)()
         if type_ in sf_custom_types:
             assert type_ == str(sample)
+
+
+@pytest.mark.feature_max_lob_size
+def test_create_table_with_text_type(engine_testaccount):
+    metadata = MetaData()
+    table_name = "test_max_lob_size_0"
+    test_max_lob_size = Table(
+        table_name,
+        metadata,
+        Column("id", Integer, primary_key=True),
+        Column("full_name", TEXT(), server_default=text("id::varchar")),
+    )
+
+    metadata.create_all(engine_testaccount)
+    try:
+        assert test_max_lob_size is not None
+
+        with engine_testaccount.connect() as conn:
+            with conn.begin():
+                query = text(f"SELECT GET_DDL('TABLE', '{table_name}')")
+                result = conn.execute(query)
+                row = str(result.mappings().fetchone())
+                assert (
+                    "VARCHAR(134217728)" in row
+                ), f"Expected VARCHAR(134217728) in {row}"
+
+    finally:
+        test_max_lob_size.drop(engine_testaccount)

tests/test_orm.py

@@ -7,6 +7,7 @@
 
 import pytest
 from sqlalchemy import (
+    TEXT,
     Column,
     Enum,
     ForeignKey,
@@ -413,3 +414,34 @@ class Employee(Base):
         '[SELECT "Employee".uid FROM "Employee" JOIN LATERAL flatten(PARSE_JSON("Employee"'
         in caplog.text
     )
+
+
+@pytest.mark.feature_max_lob_size
+def test_basic_table_with_large_lob_size_in_memory(engine_testaccount, sql_compiler):
+    Base = declarative_base()
+
+    class User(Base):
+        __tablename__ = "user"
+
+        id = Column(Integer, primary_key=True)
+        full_name = Column(TEXT(), server_default=text("id::varchar"))
+
+        def __repr__(self):
+            return f"<User({self.name!r}, {self.fullname!r})>"
+
+    Base.metadata.create_all(engine_testaccount)
+
+    try:
+        assert User.__table__ is not None
+
+        with engine_testaccount.connect() as conn:
+            with conn.begin():
+                query = text(f"SELECT GET_DDL('TABLE', '{User.__tablename__}')")
+                result = conn.execute(query)
+                row = str(result.mappings().fetchone())
+                assert (
+                    "VARCHAR(134217728)" in row
+                ), f"Expected VARCHAR(134217728) in {row}"
+
+    finally:
+        Base.metadata.drop_all(engine_testaccount)