Add "impl" parameter to PickleType

Add a impl parameter to :class:`_types.PickleType` constructor, allowing
any arbitary type to be used in place of the default implementation of
:class:`_types.LargeBinary`. Pull request courtesy jason3gb.

Fixes: #6646
Closes: #6657
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/6657
Pull-request-sha: e49bcd368d1f71dba92225d8d6e3af2bbe7142f7

Change-Id: Ib79f3b0ebbc94393f673f5a5ba6558260083d0cf

3 files changed, 39 insertions, 1 deletions

diff --git a/doc/build/changelog/unreleased_14/6646.rst b/doc/build/changelog/unreleased_14/6646.rst
new file mode 100644
index 000000000..55e0ec8d5
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/6646.rst
@@ -0,0 +1,7 @@
+.. change::
+    :tags: usecase, sql
+    :tickets: 6646
+
+    Add a impl parameter to :class:`_types.PickleType` constructor, allowing
+    any arbitary type to be used in place of the default implementation of
+    :class:`_types.LargeBinary`. Pull request courtesy jason3gb.
\ No newline at end of file
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index 4f8654afd..4ae121163 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -1783,7 +1783,11 @@ class PickleType(TypeDecorator):
     cache_ok = True
 
     def __init__(
-        self, protocol=pickle.HIGHEST_PROTOCOL, pickler=None, comparator=None
+        self,
+        protocol=pickle.HIGHEST_PROTOCOL,
+        pickler=None,
+        comparator=None,
+        impl=None,
     ):
         """
         Construct a PickleType.
@@ -1798,12 +1802,22 @@ class PickleType(TypeDecorator):
           to compare values of this type.  If left as ``None``,
           the Python "equals" operator is used to compare values.
 
+        :param impl: A binary-storing :class:`_types.TypeEngine` class or
+          instance to use in place of the default :class:`_types.LargeBinary`.
+          For example the :class: `_mysql.LONGBLOB` class may be more effective
+          when using MySQL.
+
+          .. versionadded:: 1.4.20
+
         """
         self.protocol = protocol
         self.pickler = pickler or pickle
         self.comparator = comparator
         super(PickleType, self).__init__()
 
+        if impl:
+            self.impl = to_instance(impl)
+
     def __reduce__(self):
         return PickleType, (self.protocol, None, self.comparator)
 
diff --git a/test/sql/test_types.py b/test/sql/test_types.py
index 9db0fee3b..3cbd2c07f 100644
--- a/test/sql/test_types.py
+++ b/test/sql/test_types.py
@@ -3809,6 +3809,23 @@ class PickleTest(fixtures.TestBase):
         ):
             assert p1.compare_values(p1.copy_value(obj), obj)
 
+    @testing.combinations(
+        None, mysql.LONGBLOB, LargeBinary, mysql.LONGBLOB(), LargeBinary()
+    )
+    def test_customized_impl(self, impl):
+        """test #6646"""
+
+        if impl is None:
+            p1 = PickleType()
+            assert isinstance(p1.impl, LargeBinary)
+        else:
+            p1 = PickleType(impl=impl)
+
+            if not isinstance(impl, type):
+                impl = type(impl)
+
+            assert isinstance(p1.impl, impl)
+
 
 class CallableTest(fixtures.TestBase):
     @testing.provide_metadata