Change handling of copy=None defaults for Pandas 2

sdks/python/apache_beam/dataframe/frame_base.py

@@ -678,7 +678,15 @@ def wrap(func):
     def wrapper(**kwargs):
       for name in defaults_to_populate:
         if name not in kwargs:
-          kwargs[name] = arg_to_default[name]
+          # In pandas 2, many methods rely on the default copy=None
+          # to mean that copy is the value of copy_on_write. Since
+          # copy_on_write will always be true for Beam, just fill it
+          # in here. In pandas 1, the default was True anyway.
+          if name == 'copy' and arg_to_default[name] is None:
+            kwargs[name] = True
+          else:
+            kwargs[name] = arg_to_default[name]
+
       return func(**kwargs)
 
     return wrapper

sdks/python/apache_beam/dataframe/frame_base_test.py

@@ -174,6 +174,21 @@ def func(self, a, **kwargs):
             'a': 2, 'b': 4, 'c': 6, 'kw_only': 8
         })
 
+  def test_populate_defaults_overwrites_copy(self):
+    class Base(object):
+      def func(self, a=1, b=2, c=3, *, copy=None):
+        pass
+
+    class Proxy(object):
+      @frame_base.args_to_kwargs(Base)
+      @frame_base.populate_defaults(Base)
+      def func(self, a, copy, **kwargs):
+        return dict(kwargs, a=a, copy=copy)
+
+    proxy = Proxy()
+    self.assertEqual(proxy.func(), {'a': 1, 'copy': True})
+    self.assertEqual(proxy.func(copy=False), {'a': 1, 'copy': False})
+
 
 if __name__ == '__main__':
   unittest.main()