GeezerVisonThe World Wide Web, restored to 800 pixels of glory. |
| Home|About|Gaming cabinet|Archives|How it works|Starter sites |
GeezerVison is showing github.com — feldera/python/tests/utils.py at main · feldera/feldera Expand file tree/ Copy pathutils.pyMore file actions 1066 lines (920 loc) · 41.6 KB / Copy pathutils.pyFile metadata and controls1066 lines (920 loc) · 41.6 KB 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 import contextlib import fcntl import json import logging import os import pathlib import re import shutil import subprocess import tempfile import time import uuid from collections.abc import Callable, Iterator, Sequence from dataclasses import dataclass from typing import TYPE_CHECKING from urllib.parse import urlparse from feldera.testutils import suite_tag if TYPE_CHECKING: from feldera.output_handler import OutputHandler # Environment variables naming the object stores tests use; see ObjectStore. OBJECT_STORE_URI_ENV = "CI_OBJECT_STORE_URI" INPUTS_STORE_URI_ENV = "CI_OBJECT_STORE_INPUTS_URI" # Settings of the s3:// shape of ObjectStore: a per-run MinIO, or any # S3-compatible store with static keys. Unused when OBJECT_STORE_URI_ENV names # a gs:// bucket. MINIO_ENDPOINT_ENV = "CI_MINIO_ENDPOINT" MINIO_BUCKET = os.environ.get("CI_MINIO_BUCKET", "ci-tests") MINIO_ENDPOINT = os.environ.get( MINIO_ENDPOINT_ENV, "http://minio.minio.svc.cluster.local:9000" ) MINIO_REGION = os.environ.get("CI_MINIO_REGION", "us-east-1") # rclone S3 provider name for checkpoint sync against the s3:// shape. MINIO_PROVIDER = os.environ.get("CI_MINIO_PROVIDER", "Minio") KAFKA_BOOTSTRAP = os.environ.get( "KAFKA_BOOTSTRAP_SERVERS", "ci-kafka-bootstrap.kafka:9092" ) LOCAL_FIXTURE_ROOT = pathlib.Path("/tmp/feldera_fixtures") def fixture_suite_dir() -> str: """Bucket subdirectory holding one test suite's cached fixtures. CI runs `python` and `python-multihost` concurrently against one MinIO bucket. A fixture build is not byte-reproducible (fresh `col-<uuid>` physical names, fresh Parquet file names), so two suites building at one path overwrite each other's identically-named `_delta_log/*.json` and the surviving log describes columns its data files do not have. In CI that gives `suite-m` for `python-multihost` and `suite` for `python`. """ tag = suite_tag() return f"suite-{tag}" if tag else "suite" def env_truthy(name: str) -> bool: """Return True when an environment variable is set to a truthy value.""" value = os.environ.get(name) return value is not None and value.lower() not in {"", "0", "false", "no"} def required_env(name: str) -> str: """Return a required environment variable or raise a descriptive error.""" value = os.environ.get(name) if value: return value raise RuntimeError(f"required environment variable '{name}' is not set") def runs_in_ci() -> bool: """Return True when the test suite is running under CI.""" return env_truthy("CI") def runs_on_gen2() -> bool: """Return True when ``FELDERA_RUNTIME_VERSION`` pins the test pipelines to the Gen-2 engine, which runs Rust UDFs written against the ``feldera-udf`` SDK.""" return os.environ.get("FELDERA_RUNTIME_VERSION") == "gen2" @dataclass(frozen=True) class ObjectStore: """Object store the tests keep Delta and Iceberg tables in, and how the pipeline pod and this test process authenticate to it. Two stores are named through the environment: ``OBJECT_STORE_URI_ENV`` holds per-run test data (:meth:`from_env`) and ``INPUTS_STORE_URI_ENV`` holds long-lived datasets that tests only read (:func:`inputs_store`). The URI scheme selects the shape: * ``gs://<bucket>``: Google Cloud Storage reached with the Kubernetes service account's Workload Identity on both sides. No credentials exist in the environment or in any connector config. * ``s3://<bucket>``: an S3-compatible store addressed by the ``MINIO_*`` settings above, whose static keys do end up in connector configs. """ scheme: str bucket: str # S3 only; None for GCS. endpoint: str | None = None region: str = "us-east-1" access_key: str | None = None secret_key: str | None = None @classmethod def from_env(cls) -> "ObjectStore": """The per-run test-data store: the URI in ``OBJECT_STORE_URI_ENV``, or ``s3://`` plus ``MINIO_BUCKET`` when that variable is unset.""" return cls.from_uri( os.environ.get(OBJECT_STORE_URI_ENV) or f"s3://{MINIO_BUCKET}" ) @classmethod def from_uri(cls, uri: str) -> "ObjectStore": """Parse ``<scheme>://<bucket>[/<prefix>]``; a prefix stays part of ``bucket`` so every path this store hands out lives under it.""" parsed = urlparse(uri) if parsed.scheme not in {"gs", "s3"} or not parsed.netloc: raise ValueError( "an object store URI must look like 'gs://my-bucket' or " f"'s3://my-bucket/prefix', got {uri!r}" ) bucket = f"{parsed.netloc}{parsed.path}".rstrip("/") if parsed.scheme == "gs": return cls(scheme="gs", bucket=bucket) endpoint = MINIO_ENDPOINT.rstrip("/") parsed_endpoint = urlparse(endpoint) if ( parsed_endpoint.scheme not in {"http", "https"} or not parsed_endpoint.netloc ): raise ValueError( f"{MINIO_ENDPOINT_ENV} must be a full URL, e.g. " f"'http://minio.minio.svc.cluster.local:9000', got {endpoint!r}" ) return cls( scheme="s3", bucket=bucket, endpoint=endpoint, region=MINIO_REGION, access_key=required_env("CI_K8S_MINIO_ACCESS_KEY_ID"), secret_key=required_env("CI_K8S_MINIO_SECRET_ACCESS_KEY"), ) def uri(self, path: str) -> str: return f"{self.scheme}://{self.bucket}/{path}" def delta_storage_options(self) -> dict[str, str]: """``object_store`` options shared by the Delta connector config and the ``deltalake`` package. Empty for GCS: both sides use Workload Identity.""" if self.scheme == "gs": return {} assert self.endpoint and self.access_key and self.secret_key return { "aws_access_key_id": self.access_key, "aws_secret_access_key": self.secret_key, "aws_region": self.region, "aws_endpoint": self.endpoint, "aws_allow_http": str(urlparse(self.endpoint).scheme == "http").lower(), } def iceberg_fileio_config(self) -> dict[str, str]: """``pyiceberg`` FileIO properties, also passed to the Iceberg connector. Empty for GCS for the same reason as :meth:`delta_storage_options`.""" if self.scheme == "gs": return {} assert self.endpoint and self.access_key and self.secret_key return { "s3.endpoint": self.endpoint, "s3.access-key-id": self.access_key, "s3.secret-access-key": self.secret_key, "s3.region": self.region, "s3.path-style-access": "true", } def sync_bucket(self, bucket_path: str) -> str: """``bucket`` value of a checkpoint-sync config for ``<bucket>/<prefix>``. Checkpoint sync selects rclone's backend by the scheme: ``gs://`` is Google Cloud Storage over its own API, anything else is S3. """ return f"gs://{bucket_path}" if self.scheme == "gs" else bucket_path def checkpoint_sync_connection( self, bucket_path: str, *, auth_err: bool = False, endpoint: str | None = None, ) -> dict: """Connection half of a pipeline's checkpoint-sync ``sync`` config. ``auth_err`` breaks authentication on purpose: anonymous requests for GCS (denied on a private bucket), a corrupted secret key for S3. """ sync: dict = {"bucket": self.sync_bucket(bucket_path)} if self.scheme == "gs": if endpoint is not None: sync["endpoint"] = endpoint if auth_err: # rclone honors `anonymous` ahead of `env_auth`. sync["flags"] = ["--gcs-anonymous"] return sync assert self.endpoint and self.access_key and self.secret_key sync.update( { "access_key": self.access_key, "secret_key": self.secret_key + ("extra" if auth_err else ""), "provider": MINIO_PROVIDER, "endpoint": endpoint or self.endpoint, "region": self.region, } ) return sync @property def auth_error_pattern(self) -> str: """Regex matching the error a pipeline reports for ``auth_err`` syncs.""" if self.scheme == "gs": # Anchored: bare digits would also match UUIDs, ports and sizes. return r"Anonymous caller|Forbidden|\bError 40[13]\b" return "SignatureDoesNotMatch|Forbidden" def pyarrow_fs(self): """pyarrow filesystem for this store; paths are ``<bucket>/<key>``. Imported lazily to keep test collection cheap on hosts that never touch object storage. """ import pyarrow.fs as pafs if self.scheme == "gs": from datetime import timedelta # Credentials come from the GKE metadata server (Workload Identity). # Without one the client retries indefinitely; bounding it lets # callers fall back instead of hanging the run. return pafs.GcsFileSystem(retry_time_limit=timedelta(seconds=30)) assert self.endpoint parsed_endpoint = urlparse(self.endpoint) return pafs.S3FileSystem( access_key=self.access_key, secret_key=self.secret_key, region=self.region, scheme=parsed_endpoint.scheme, endpoint_override=parsed_endpoint.netloc, ) def inputs_store() -> ObjectStore | None: """Store holding long-lived datasets the tests only read, or None where the CI environment has none (the fresh-install CI runs a per-run MinIO only).""" uri = os.environ.get(INPUTS_STORE_URI_ENV) return ObjectStore.from_uri(uri) if uri else None @dataclass class DeltaTestLocation: """Describe where the Delta sink writes test data and how to read it back.""" uri: str connector_config: dict[str, object] root_path: str local_dir: pathlib.Path | None = None # True when ``stable_subpath`` was used at construction time. ``cleanup()`` # honors this by leaving the directory in place so the next run reuses the # cached fixture instead of paying to rebuild it. stable: bool = False # The remote store behind ``uri``; None for a local directory. store: ObjectStore | None = None # Written last by ``_place_tree``: a builder killed mid-upload leaves a # tree that already satisfies ``delta_log_exists``. READY_MARKER = "_fixture_ready" @classmethod def create( cls, pipeline_name: str, *, mode: str = "truncate", stable_subpath: str | None = None, ) -> "DeltaTestLocation": """Use the local filesystem for local runs and :class:`ObjectStore` in CI. :param mode: Value of the connector's ``mode`` field. Output connectors use ``"truncate"`` (the default); input connectors should pass ``"snapshot"``. :param stable_subpath: When set, locates the table at a fixed path (``_fixtures/<suite>/<stable_subpath>``) that is *not* namespaced by the pipeline name or commit SHA, so every test needing that fixture shares one build of it. When unset, a fresh random subpath under ``pipeline_name`` is used (the original behavior). Use a stable path only for fixtures whose contents are deterministic across runs. The cache spans runs locally, but not in CI: MinIO and its volume are created fresh per integration-test run, so each suite rebuilds every fixture once per run. See :func:`fixture_suite_dir` for why suites must not share the path. """ if runs_in_ci(): store = ObjectStore.from_env() if stable_subpath is not None: prefix = f"_fixtures/{fixture_suite_dir()}/{stable_subpath}" else: prefix = f"{pipeline_name}/{uuid.uuid4().hex}" return cls( uri=store.uri(prefix), connector_config={ "uri": store.uri(prefix), "mode": mode, **store.delta_storage_options(), }, root_path=f"{store.bucket}/{prefix}", stable=stable_subpath is not None, store=store, ) if stable_subpath is not None: # No suite split: one suite per developer machine. local_dir = LOCAL_FIXTURE_ROOT / stable_subpath local_dir.mkdir(parents=True, exist_ok=True) else: local_dir = pathlib.Path( tempfile.mkdtemp(prefix=f"{pipeline_name}_delta_", dir="/tmp") ) return cls( uri=f"file://{local_dir}", connector_config={ "uri": f"file://{local_dir}", "mode": mode, }, root_path=str(local_dir), local_dir=local_dir, stable=stable_subpath is not None, ) @classmethod def dataset( cls, relative_path: str, *, mode: str = "snapshot" ) -> "DeltaTestLocation": """A read-only Delta dataset, addressed relative to the inputs store. In CI the dataset lives in the store named by ``INPUTS_STORE_URI_ENV``; a run without that store cannot serve it, so callers should skip on the ``RuntimeError``. Locally, datasets are read from ``LOCAL_FIXTURE_ROOT/datasets/<relative_path>``. """ if runs_in_ci(): store = inputs_store() if store is None: raise RuntimeError( f"{INPUTS_STORE_URI_ENV} is not set: this environment has " f"no read-only datasets (wanted {relative_path!r})" ) return cls( uri=store.uri(relative_path), connector_config={ "uri": store.uri(relative_path), "mode": mode, **store.delta_storage_options(), }, root_path=f"{store.bucket}/{relative_path}", stable=True, store=store, ) local_dir = LOCAL_FIXTURE_ROOT / "datasets" / relative_path return cls( uri=f"file://{local_dir}", connector_config={"uri": f"file://{local_dir}", "mode": mode}, root_path=str(local_dir), local_dir=local_dir, stable=True, ) def delta_storage_options(self) -> dict[str, str]: """Return `deltalake` storage_options derived from the connector config.""" return { k: str(v) for k, v in self.connector_config.items() if k not in ("uri", "mode") } def writer_storage_options(self) -> dict[str, str] | None: """`storage_options` for `deltalake.write_deltalake`, or None for local. S3 writes need the unsafe-rename opt-in: deltalake's default lock provider is DynamoDB, which the test environment does not run. """ opts = self.delta_storage_options() if not opts: return None if self.uri.startswith("s3://"): opts.setdefault("aws_s3_allow_unsafe_rename", "true") return opts def _filesystem(self): """pyarrow filesystem of the remote store behind this location.""" assert self.store is not None, "no remote store behind a local location" return self.store.pyarrow_fs() def log_json_paths(self) -> list[str]: """List Delta transaction log JSON files in version order.""" if self.local_dir is not None: return [ str(path) for path in sorted((self.local_dir / "_delta_log").glob("*.json")) ] import pyarrow.fs as pafs fs = self._filesystem() infos = fs.get_file_info( pafs.FileSelector(f"{self.root_path}/_delta_log", recursive=False) ) return sorted( info.path for info in infos if info.type == pafs.FileType.File and info.path.endswith(".json") ) def _read_text(self, path: str) -> str: """Read a text file from the configured backend.""" if self.local_dir is not None: return pathlib.Path(path).read_text(encoding="utf-8") with self._filesystem().open_input_file(path) as handle: return handle.readall().decode("utf-8") def _read_parquet(self, relative_path: str): """Read a Delta data file (relative to ``root_path``) as a pyarrow.Table.""" import pyarrow.parquet as pq if self.local_dir is not None: return pq.read_table(self.local_dir / relative_path) return pq.read_table( f"{self.root_path}/{relative_path}", filesystem=self._filesystem() ) def read_rows(self) -> list[dict]: """Read the active rows of the Delta table by replaying its log. Walks ``_delta_log/*.json``, follows ``add``/``remove`` actions to derive the current set of parquet files, reads them with pyarrow and drops Feldera-internal ``__feldera_*`` columns. Works against both local-filesystem and S3-backed (MinIO) tables; avoids the ``deltalake`` Python package, whose wheel aborts on aarch64 hosts. """ import pyarrow as pa active: dict[str, None] = {} for log_path in self.log_json_paths(): for line in self._read_text(log_path).splitlines(): action = json.loads(line) if (add := action.get("add")) is not None: active[add["path"]] = None if (remove := action.get("remove")) is not None: active.pop(remove["path"], None) if not active: return [] tables = [self._read_parquet(rel) for rel in sorted(active)] return [ { key: value for key, value in row.items() if not key.startswith("__feldera_") } for row in pa.concat_tables(tables).to_pylist() ] def live_row_count(self) -> int: """Rows a reader sees, counting deletion vectors. Replays ``_delta_log/*.json`` to find the current ``add`` per path, then sums each file's ``numRecords`` less the cardinality of its deletion vector. That is the arithmetic the Delta protocol defines for a live row count, and it is what :meth:`row_count` gets wrong on a table written by the merge-mode output connector, which supersedes rows with vectors rather than by rewriting files. Reads the log rather than the data because no Python Delta reader available here applies deletion vectors: the pinned ``deltalake`` wheel refuses a table that advertises the ``deletionVectors`` reader feature outright. Row-level cross-engine checks live in the Rust tests, which drive Delta Spark. """ # Keyed by path: within one commit the connector emits the `remove` and # then the `add` for a file whose vector changed, so last write wins and # leaves the current `add`. active: dict[str, dict] = {} for log_path in self.log_json_paths(): for line in self._read_text(log_path).splitlines(): action = json.loads(line) if (add := action.get("add")) is not None: active[add["path"]] = add elif (remove := action.get("remove")) is not None: active.pop(remove["path"], None) total = 0 for add in active.values(): stats = add.get("stats") if stats is None: raise AssertionError( f"data file {add['path']!r} has no statistics, so its rows " "cannot be counted from the log" ) rows = json.loads(stats)["numRecords"] vector = add.get("deletionVector") total += rows - (vector["cardinality"] if vector else 0) return total def row_count(self, missing_ok: bool = False) -> int: """Return the row count of the current Delta snapshot. Uses the per-file `numRecords` stats recorded in the delta log, so we never need to scan parquet (and never need pyarrow). The `deltalake` import is deferred to keep module-level test collection cheap on hosts that don't read delta tables. :param missing_ok: When True, return ``-1`` if the table does not exist instead of raising. Useful when probing a cache. """ from deltalake import DeltaTable from deltalake.exceptions import TableNotFoundError try: dt = DeltaTable(self.uri, storage_options=self.delta_storage_options()) except TableNotFoundError: if missing_ok: return -1 raise return dt.count() def delta_log_exists(self) -> bool: """Return True when a Delta log is present at this location. Used to decide whether a cached fixture (see ``stable_subpath``) can be reused instead of rebuilt. """ try: return len(self.log_json_paths()) > 0 except FileNotFoundError: return False def fixture_is_complete(self) -> bool: """Return True when a fully placed fixture tree is present here. A half-placed tree lacks the ``READY_MARKER`` and so reads as absent. """ if self.local_dir is not None: return (self.local_dir / self.READY_MARKER).is_file() import pyarrow.fs as pafs info = self._filesystem().get_file_info(f"{self.root_path}/{self.READY_MARKER}") return info.type == pafs.FileType.File def _place_tree(self, staging: pathlib.Path) -> None: """Copy a locally-built Delta table tree to where this location stores its data: the local directory, or the S3/MinIO bucket, depending on how this location was created. Some fixtures can only be produced on the local filesystem (e.g. a PySpark-written table). For a local target, any existing content at ``self.local_dir`` is deleted before the copy. S3/MinIO targets get the data files first and ``_delta_log`` last, so a reader observing the upload mid-flight never sees a log referencing a missing parquet. The ``READY_MARKER`` is removed before the first write and written last of all, so a reader that finds it knows the whole tree landed. """ if self.local_dir is not None: if self.local_dir.exists(): shutil.rmtree(self.local_dir) # `dirs_exist_ok`: an unlocked `create()` elsewhere may have # recreated the directory between the rmtree and here. shutil.copytree(staging, self.local_dir, dirs_exist_ok=True) (self.local_dir / self.READY_MARKER).write_text("", encoding="utf-8") return fs = self._filesystem() # Marker off before the first byte lands, on after the last, so a # reader that finds it never sees a half-replaced tree. with contextlib.suppress(FileNotFoundError): fs.delete_file(f"{self.root_path}/{self.READY_MARKER}") files = [path for path in sorted(staging.rglob("*")) if path.is_file()] for path in sorted(files, key=lambda p: ("_delta_log" in p.parts, p.name)): rel = path.relative_to(staging).as_posix() with fs.open_output_stream(f"{self.root_path}/{rel}") as out: out.write(path.read_bytes()) with fs.open_output_stream(f"{self.root_path}/{self.READY_MARKER}") as out: out.write(b"") def fetch_tree(self, dest: pathlib.Path) -> pathlib.Path: """Copy this location's Delta table to a local directory and return it. The inverse of :meth:`_place_tree`, for a tool that speaks only the local filesystem. Spark reading S3 would need an S3A stack on top of the Delta JARs, and the bytes are the same either way. """ dest.mkdir(parents=True, exist_ok=True) if self.local_dir is not None: shutil.copytree(self.local_dir, dest, dirs_exist_ok=True) return dest import pyarrow.fs as pafs fs = self._filesystem() infos = fs.get_file_info(pafs.FileSelector(self.root_path, recursive=True)) for info in infos: if info.type != pafs.FileType.File: continue target = dest / pathlib.PurePosixPath(info.path).relative_to(self.root_path) target.parent.mkdir(parents=True, exist_ok=True) with fs.open_input_file(info.path) as handle: target.write_bytes(handle.readall()) return dest def cleanup(self) -> None: """Remove the local temp directory, if any. No-op when ``stable_subpath`` was used at construction time: the whole point of a stable path is to cache contents across runs, so deleting it would defeat the cache. Also a no-op on the CI/MinIO path: there is no local directory to remove, so any objects this run wrote to the bucket are simply left in place. The shared MinIO bucket is long-lived and these tests never delete from it, so those objects accumulate; non-stable runs each use a unique uuid prefix so they never collide, and the leftover volume for this internal CI bucket is accepted rather than swept here. """ if self.local_dir is not None and not self.stable: shutil.rmtree(self.local_dir, ignore_errors=True) self.local_dir = None @dataclass class IcebergTestLocation: """Where an Iceberg test table lives and how to read/write it. A test writes rows with ``pyiceberg`` (the ``iceberg-rust`` connector cannot write yet) and then points the connector at the table via ``metadata_location`` — the catalog-free access path. Both sides reach the same storage: * Local runs put the warehouse under ``/tmp`` and use bare filesystem paths (no ``file://`` scheme), matching the Rust FS harness and the fork's local ``FileIO`` resolver. * CI runs put the warehouse in the in-cluster MinIO bucket over S3, so the pipeline pod and the test runner reach the table through the same object store, exactly like :class:`DeltaTestLocation`. The ``pyiceberg`` SQL catalog metadata (a SQLite file) always lives on local disk; it is used only by the writer (the test runner), never by the connector. """ warehouse: str table_location: str catalog_db: str namespace: str table_name: str fileio_config: dict[str, str] # Local directory removed on teardown. In local runs it is the whole # warehouse (data + SQLite catalog); in CI it is only the catalog dir (the # warehouse lives in S3/MinIO). `None` leaves nothing to remove. cleanup_dir: pathlib.Path | None = None NAMESPACE = "iceberg_test" TABLE = "test_table" @classmethod def create(cls, pipeline_name: str) -> "IcebergTestLocation": """Local filesystem for local runs, :class:`ObjectStore` in CI.""" if runs_in_ci(): store = ObjectStore.from_env() prefix = f"{pipeline_name}/{uuid.uuid4().hex}" fileio_config = store.iceberg_fileio_config() # The SQLite catalog is only touched by the writer, so it stays # on local disk even when the warehouse is remote. catalog_dir = pathlib.Path( tempfile.mkdtemp(prefix=f"{pipeline_name}_iceberg_cat_", dir="/tmp") ) return cls( warehouse=store.uri(prefix), table_location=store.uri(f"{prefix}/{cls.TABLE}"), catalog_db=str(catalog_dir / "catalog.db"), namespace=cls.NAMESPACE, table_name=cls.TABLE, fileio_config=fileio_config, cleanup_dir=catalog_dir, ) local_dir = pathlib.Path( tempfile.mkdtemp(prefix=f"{pipeline_name}_iceberg_", dir="/tmp") ) # Bare (scheme-less) table path: the Rust FS harness reads tables # written this way, and the connector's local resolver expects it. return cls( warehouse=f"file://{local_dir}", table_location=f"{local_dir}/{cls.TABLE}", catalog_db=str(local_dir / "catalog.db"), namespace=cls.NAMESPACE, table_name=cls.TABLE, fileio_config={}, cleanup_dir=local_dir, ) @property def qualified_name(self) -> str: return f"{self.namespace}.{self.table_name}" def _catalog(self): """Build the ``pyiceberg`` SQL catalog. Deferred import keeps module collection cheap on hosts that never touch Iceberg.""" from pyiceberg.catalog.sql import SqlCatalog return SqlCatalog( "test", **{ "uri": f"sqlite:///{self.catalog_db}", "warehouse": self.warehouse, **self.fileio_config, }, ) def create_table(self, schema, partition_spec=None): """Create (replacing any prior) the test table and return it.""" catalog = self._catalog() try: catalog.create_namespace(self.namespace) except Exception: pass # Already exists. try: catalog.drop_table(self.qualified_name) except Exception: pass # Nothing to drop. from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC return catalog.create_table( self.qualified_name, schema, location=self.table_location, partition_spec=partition_spec or UNPARTITIONED_PARTITION_SPEC, ) def append(self, arrow_table) -> None: """Append one Arrow batch as a new Iceberg snapshot.""" table = self._catalog().load_table(self.qualified_name) table.append(arrow_table) def metadata_location(self) -> str: """Return the current table metadata file location. This is the value handed to the connector's ``metadata_location`` option, so a reload picks up the latest snapshot after appends. """ return self._catalog().load_table(self.qualified_name).metadata_location def connector_config(self, **extra) -> dict[str, object]: """Transport config pointing the connector at this table. Merges ``metadata_location`` and any storage (``s3.*``) options with caller-supplied ``extra`` fields (e.g. ``mode``, ``timestamp_column``). """ config: dict[str, object] = {"metadata_location": self.metadata_location()} config.update(self.fileio_config) config.update(extra) return config def row_count(self) -> int: table = self._catalog().load_table(self.qualified_name) return table.scan().to_arrow().num_rows def remove_if_local(self) -> None: """Remove the local temp directory, if any. A no-op on the S3 data itself (as with :class:`DeltaTestLocation`): objects this run wrote to the shared MinIO bucket use a unique prefix and are left in place. In CI only the local SQLite catalog dir is removed; in local runs the whole warehouse dir is removed. """ if self.cleanup_dir is not None: shutil.rmtree(self.cleanup_dir, ignore_errors=True) self.cleanup_dir = None @contextlib.contextmanager def _fixture_build_lock(key: str) -> Iterator[None]: """Serialize the check-then-build of the fixture at ``key`` across processes. :func:`fixture_suite_dir` keeps the other suite off this path, so every contender is an xdist worker on this host and a local ``flock`` suffices even for a remote fixture. The kernel drops it if a builder dies. """ lock_dir = LOCAL_FIXTURE_ROOT / ".locks" lock_dir.mkdir(parents=True, exist_ok=True) lock_path = lock_dir / f"{re.sub(r'[^A-Za-z0-9._-]', '_', key)}.lock" with open(lock_path, "w") as handle: fcntl.flock(handle, fcntl.LOCK_EX) try: yield finally: fcntl.flock(handle, fcntl.LOCK_UN) def run_delta_spark( script: str | os.PathLike[str], args: Sequence[object] = (), *, delta_spark_spec: str = "delta-spark>=4.2,<5", max_attempts: int = 3, capture_output: bool = True, ) -> str: """Run a standalone Delta Spark script and return its stdout. Runs under ``uv run --no-project --with <delta_spark_spec>``, so the PySpark and JVM stack is pulled only when a test needs it. Retries because Spark resolves delta-spark from Maven Central here, which fails transiently. :param args: Positional arguments after the script path, each stringified. :param capture_output: ``False`` lets output through to the test log. """ if shutil.which("uv") is None: raise RuntimeError( "`uv` is required on PATH to run a Delta Spark script " f"(runs via `uv run --with {delta_spark_spec}`)." ) command = [ "uv", "run", "--no-project", "--with", delta_spark_spec, "python", str(script), *(str(arg) for arg in args), ] for attempt in range(1, max_attempts + 1): try: # Every run shares uv's package cache and Spark's Ivy cache, and # neither tolerates concurrent writers: uv times out on its lock # and Ivy leaves truncated jars. Runs are rare, so serialize them. with _fixture_build_lock("delta-spark"): completed = subprocess.run( command, check=True, capture_output=capture_output, text=True ) return completed.stdout if capture_output else "" except subprocess.CalledProcessError as e: if attempt == max_attempts: raise logging.warning( "Delta Spark script %s failed (attempt %d/%d); retrying. " "Usually a transient Maven/Ivy download failure.\n%s", script, attempt, max_attempts, e.stderr, ) time.sleep(5 * attempt) raise AssertionError("unreachable") def ensure_delta_spark_fixture( loc: DeltaTestLocation, builder_script: str | os.PathLike[str], builder_args: Sequence[object] = (), *, delta_spark_spec: str = "delta-spark>=4.2,<5", is_present: Callable[[DeltaTestLocation], bool] | None = None, max_attempts: int = 3, ) -> None: """Ensure a PySpark-authored Delta fixture exists at ``loc`` (cached). Some Delta features (deletion vectors, column-mapping schema evolution) can only be written by Delta Spark, not by delta-rs or the ``deltalake`` wheel. This builds such a fixture once and reuses it: * If the fixture is already present (the completion marker is there and ``is_present``, defaulting to :meth:`DeltaTestLocation.delta_log_exists`, agrees), do nothing, so every test needing it shares one build. * Otherwise run ``builder_script`` in an isolated environment (``uv run --no-project --with <delta_spark_spec> python <builder_script> <staging> *builder_args``), staging into a temp dir so a half-finished build can never leak into the upload, then place the tree onto ``loc``'s backend. ``--no-project`` keeps the builder hermetic: it depends only on ``delta_spark_spec``, never on building the enclosing project. The heavy PySpark + JVM stack is pulled only on this rare rebuild path. :param builder_script: Path to a standalone script that writes a Delta table to the directory given as its first argument. :param builder_args: Extra positional arguments passed after the staging directory. Each is stringified verbatim with ``str()`` — pass primitives; ``None`` would become the literal string ``"None"``. :param is_present: Predicate deciding whether the fixture already holds the content this caller wants; also re-checked after upload. It is ANDed with :meth:`DeltaTestLocation.fixture_is_complete`, which is what rules out a partially placed tree. :param max_attempts: How many times to run the builder before giving up. Spark resolves the delta-spark dependency tree from Maven Central on this path; that download is prone to transient failures (0-byte artifacts, half-written Ivy cache files, gateway timeouts) that clear on a fresh attempt, so retry rather than fail the test on infra flakiness. """ matches = ( is_present if is_present is not None else DeltaTestLocation.delta_log_exists ) def present(location: DeltaTestLocation) -> bool: return location.fixture_is_complete() and matches(location) if present(loc): return if shutil.which("uv") is None: raise RuntimeError( "`uv` is required on PATH to build the PySpark Delta fixture " f"(builder runs via `uv run --with {delta_spark_spec}`)." ) with _fixture_build_lock(loc.root_path): _build_delta_spark_fixture( loc, builder_script, builder_args, delta_spark_spec=delta_spark_spec, present=present, max_attempts=max_attempts, ) def _build_delta_spark_fixture( loc: DeltaTestLocation, builder_script: str | os.PathLike[str], builder_args: Sequence[object], *, delta_spark_spec: str, present: Callable[[DeltaTestLocation], bool], max_attempts: int, ) -> None: """Build and place the fixture at ``loc``. Call under the build lock.""" # Whoever held the lock before us may have built it already. if present(loc): return for attempt in range(1, max_attempts + 1): # Fresh staging each attempt: a builder that died mid-write must not # leak a partial tree into the next attempt or the upload. staging = pathlib.Path(tempfile.mkdtemp(prefix="feldera_delta_fixture_")) try: run_delta_spark( builder_script, [staging, *builder_args], delta_spark_spec=delta_spark_spec, max_attempts=1, capture_output=False, ) loc._place_tree(staging) break except subprocess.CalledProcessError: if attempt == max_attempts: raise logging.warning( "PySpark Delta fixture build failed (attempt %d/%d); retrying. " "Usually a transient Maven/Ivy download failure.", attempt, max_attempts, ) time.sleep(5 * attempt) finally: shutil.rmtree(staging, ignore_errors=True) if not present(loc): raise RuntimeError( f"Delta fixture at {loc.uri} is still absent after the builder " "ran and the tree was uploaded." ) def wait_for_condition( description: str, predicate_func, timeout_s: float | None, poll_interval_s: float, ) -> None: """Poll ``predicate_func`` until it returns truthy or the timeout elapses. |
| You are visitor 01402503 — GeezerVison does not store the pages you look up. Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required. |