SNOW-1934035 add support for file (#2177)

Author: sfc-gh-zli

src/snowflake/connector/constants.py

@@ -182,6 +182,9 @@ def struct_pa_type(metadata: ResultMetadataV2) -> DataType:
     ),
     FieldType(name="VECTOR", dbapi_type=[DBAPI_TYPE_BINARY], pa_type=vector_pa_type),
     FieldType(name="MAP", dbapi_type=[DBAPI_TYPE_BINARY], pa_type=map_pa_type),
+    FieldType(
+        name="FILE", dbapi_type=[DBAPI_TYPE_STRING], pa_type=lambda _: pa.string()
+    ),
 )
 
 FIELD_NAME_TO_ID: DefaultDict[Any, int] = defaultdict(int)

test/integ/test_cursor.py

@@ -724,6 +724,46 @@ def test_geometry(conn_cnx):
                 assert row in expected_data
 
 
+@pytest.mark.skipolddriver
+def test_file(conn_cnx):
+    """Variant including JSON object."""
+    name_file = random_string(5, "test_file_")
+    with conn_cnx(
+        session_parameters={
+            "ENABLE_FILE_DATA_TYPE": True,
+        },
+    ) as cnx:
+        with cnx.cursor() as cur:
+            cur.execute(
+                f"create temporary table {name_file} as select "
+                f"TO_FILE(OBJECT_CONSTRUCT('RELATIVE_PATH', 'some_new_file.jpeg', 'STAGE', '@myStage', "
+                f"'STAGE_FILE_URL', 'some_new_file.jpeg', 'SIZE', 123, 'ETAG', 'xxx', 'CONTENT_TYPE', 'image/jpeg', "
+                f"'LAST_MODIFIED', '2025-01-01')) as file_col"
+            )
+
+            expected_data = [
+                {
+                    "RELATIVE_PATH": "some_new_file.jpeg",
+                    "STAGE": "@myStage",
+                    "STAGE_FILE_URL": "some_new_file.jpeg",
+                    "SIZE": 123,
+                    "ETAG": "xxx",
+                    "CONTENT_TYPE": "image/jpeg",
+                    "LAST_MODIFIED": "2025-01-01",
+                }
+            ]
+
+        with cnx.cursor() as cur:
+            # Test with FILE return type
+            result = cur.execute(f"select * from {name_file}")
+            for metadata in [cur.description, cur._description_internal]:
+                assert FIELD_ID_TO_NAME[metadata[0].type_code] == "FILE"
+            data = result.fetchall()
+            for raw_data in data:
+                row = json.loads(raw_data[0])
+                assert row in expected_data
+
+
 @pytest.mark.skipolddriver
 def test_vector(conn_cnx, is_public_test):
     if is_public_test: