[python] Make DataSketches optional for regular reads

.github/workflows/paimon-python-checks.yml

@@ -126,7 +126,7 @@ jobs:
             # ray/lance/daft/torch have no 3.7 wheels; those tests are importorskip-ed (as on 3.6.15).
             python -m pip install --upgrade pip
             python -m pip install -r paimon-python/dev/requirements.txt
-            python -m pip install flake8==4.0.1 'pytest~=7.0' py4j==0.10.9.9 parameterized==0.9.0
+            python -m pip install flake8==4.0.1 'pytest~=7.0' py4j==0.10.9.9 parameterized==0.9.0 datasketches==4.1.0
             python -m pip install 'lumina-data>=${{ env.LUMINA_DATA_VERSION }}' -i https://pypi.org/simple/
           else
             python -m pip install --upgrade pip

paimon-python/dev/requirements.txt

@@ -42,5 +42,3 @@ zstandard>=0.19,<1
 backports.zstd>=1.0.0,<1.4.0; python_version >= "3.9" and python_version < "3.14"
 cramjam>=1.3.0,<3; python_version>="3.7"
 pyyaml>=5.4,<7
-datasketches>=4,<5; python_version < "3.9"
-datasketches>=5,<6; python_version >= "3.9"

paimon-python/pypaimon/read/reader/aggregate/aggregators.py

@@ -34,8 +34,6 @@
 """
 from typing import Any, List, Dict, Optional, Tuple, Union, Set
 
-from _datasketches import compact_theta_sketch, theta_union
-
 from pypaimon.common.options import CoreOptions
 from pypaimon.common.options.core_options import NestedKeyNullStrategy
 from pypaimon.data.decimal import Decimal
@@ -1414,6 +1412,15 @@ def agg(self, accumulator: Any, input_field: Any) -> Any:
         if isinstance(input_field, bytearray):
             input_field = bytes(input_field)
 
+        try:
+            from _datasketches import compact_theta_sketch, theta_union
+        except ImportError as exc:
+            raise ImportError(
+                "The theta_sketch aggregator requires the 'datasketches' "
+                "package. Install it with "
+                "\"pip install 'pypaimon[theta-sketch]'\"."
+            ) from exc
+
         sketch1 = compact_theta_sketch.deserialize(accumulator)
         sketch2 = compact_theta_sketch.deserialize(input_field)
 

paimon-python/pypaimon/tests/test_optional_datasketches_dependency.py

@@ -0,0 +1,91 @@
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import os
+import subprocess
+import sys
+import textwrap
+import unittest
+
+
+class OptionalDataSketchesDependencyTest(unittest.TestCase):
+
+    def test_missing_datasketches_does_not_block_regular_aggregators(self):
+        code = textwrap.dedent(
+            """
+            import builtins
+
+            real_import = builtins.__import__
+
+            def import_without_datasketches(
+                    name, globals=None, locals=None, fromlist=(), level=0):
+                if name == "_datasketches":
+                    raise ModuleNotFoundError(
+                        "No module named '_datasketches'")
+                return real_import(
+                    name, globals, locals, fromlist, level)
+
+            builtins.__import__ = import_without_datasketches
+
+            import pypaimon
+            from pypaimon.common.options import CoreOptions, Options
+            from pypaimon.read.reader.aggregate import create_field_aggregator
+            from pypaimon.schema.data_types import AtomicType
+
+            options = CoreOptions(Options.from_none())
+            sum_agg = create_field_aggregator(
+                AtomicType("INT"), "value", "sum", options)
+            assert sum_agg.agg(1, 2) == 3
+
+            theta_agg = create_field_aggregator(
+                AtomicType("VARBINARY"), "value", "theta_sketch", options)
+            try:
+                theta_agg.agg(b"first", b"second")
+            except ImportError as exc:
+                assert "pypaimon[theta-sketch]" in str(exc)
+            else:
+                raise AssertionError(
+                    "theta_sketch should require datasketches")
+            """
+        )
+        env = os.environ.copy()
+        python_root = os.path.abspath(
+            os.path.join(os.path.dirname(__file__), "..", ".."))
+        existing = env.get("PYTHONPATH")
+        env["PYTHONPATH"] = (
+            python_root
+            if not existing
+            else python_root + os.pathsep + existing
+        )
+        result = subprocess.run(
+            [sys.executable, "-c", code],
+            env=env,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            universal_newlines=True,
+        )
+        self.assertEqual(
+            0,
+            result.returncode,
+            "subprocess failed:\nstdout:\n{}\nstderr:\n{}".format(
+                result.stdout, result.stderr),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

paimon-python/setup.py

@@ -186,6 +186,10 @@ def read_requirements():
         'full-text': [
             'paimon-ftindex==0.1.0; python_version>="3.8"',
         ],
+        'theta-sketch': [
+            'datasketches>=4,<5; python_version<"3.9"',
+            'datasketches>=5,<6; python_version>="3.9"',
+        ],
         'sql': [
             'pypaimon-rust>=0.3.0; python_version>="3.10"',
             'datafusion>=54,<55; python_version>="3.10"',