diff --git a/docs/dqx/docs/installation.mdx b/docs/dqx/docs/installation.mdx index 3a2f56b60..7f7f16f9b 100644 --- a/docs/dqx/docs/installation.mdx +++ b/docs/dqx/docs/installation.mdx @@ -66,6 +66,14 @@ pip install 'databricks-labs-dqx[pii]' This installs additional dependencies including Microsoft Presidio library. +### Installing DQX with company-hosted PyPI mirror + +Some enterprises block the public PyPI index and host a company-controlled PyPI mirror. To install DQX while using a company-hosted PyPI mirror for finding its dependencies, add all DQX dependencies to the company-hosted PyPI mirror (see "dependencies" in [`pyproject.toml`](https://github.com/databrickslabs/dqx/blob/v0.10.0/pyproject.toml)) and set the environment variable `PIP_INDEX_URL` to the company-hosted PyPI mirror URL while installing DQX: + +```commandline +PIP_INDEX_URL="https://url-to-company-hosted-pypi.internal" pip install databricks-labs-dqx +``` + ## DQX installation as a Tool in a Databricks Workspace If you install DQX via PyPI and use it purely as a library, you don’t need to pre-install DQX in the workspace. @@ -358,6 +366,26 @@ resources: # package: databricks-labs-dqx==0.8.0 ``` +### Installing DQX with company-hosted PyPI mirror + +Some enterprises block the public PyPI index and host a company-controlled PyPI mirror. To install DQX while using a company-hosted PyPI mirror for finding its dependencies, add all DQX dependencies to the company-hosted PyPI mirror (see "dependencies" in [`pyproject.toml`](https://github.com/databrickslabs/dqx/blob/v0.10.0/pyproject.toml)) and set the environment variable `PIP_INDEX_URL` to the company-hosted PyPI mirror URL while installing DQX: + +```commandline +PIP_INDEX_URL="https://url-to-company-hosted-pypi.internal" databricks labs install dqx +``` + +During DQX installation as a tool in the workspace reply *yes* to the question "Does the given workspace block Internet access"? + +If the installation host has no access to GitHub, then dqx installation will not be able to download the files locally and will fail. +In order to address that, follow the steps below + - install dqx on a host which has access to github using the instruction above + - zip the installation from ~/.databricks/labs/dqx + - copy the zip file to the target host and unzip +Now the installation can be done in offline mode (ensure Databricks CLI is upgraded to version v0.244.0 or higher) +```commandline +PIP_INDEX_URL="https://url-to-company-hosted-pypi.internal" databricks labs install dqx --offline=true +``` + ### Upgrade DQX in the Databricks workspace Verify that DQX is installed: diff --git a/docs/dqx/docs/reference/benchmarks.mdx b/docs/dqx/docs/reference/benchmarks.mdx index 56a408ce0..5bddeb18e 100644 --- a/docs/dqx/docs/reference/benchmarks.mdx +++ b/docs/dqx/docs/reference/benchmarks.mdx @@ -9,9 +9,11 @@ sidebar_position: 13 # Performance Benchmarks Report ## Specification + * 100 million rows are used for each test. -* Tests are run sequentially to reduce variability. +* DQX rules are executed in parallel and in distributed manner using Spark. To minimize test variability, each type of check is executed sequentially in this benchmark. Specific tests, such as `test_benchmark_apply_checks_all_dataset_checks` and `test_benchmark_apply_checks_all_row_checks`, execute all types of checks in parallel simultaneously. * Benchmarks are created using Databricks Serverless cluster. +* The provided benchmarks are indicative. You should always consider benchmarking results in the context of your own data and environment. ## Results | Test | Mean (s) | Median (s) | Min (s) | Max (s) | Stddev (s) | iqr (s) | q1 (s) | q3 (s) | Rounds | iqr outliers | stddev outliers | Ops/s | diff --git a/pyproject.toml b/pyproject.toml index 0f43c92f2..ad30a92cf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,10 +34,12 @@ classifiers = [ "Topic :: Software Development :: Libraries", "Topic :: Utilities", ] -dependencies = ["databricks-labs-blueprint>=0.9.1,<0.10", - "databricks-sdk~=0.71", - "databricks-labs-lsql>=0.5,<=0.16", - "sqlalchemy>=1.4,<3.0", +# also update fallback dependencies in workflow installer when changing dependencies! +dependencies = [ + "databricks-labs-blueprint>=0.9.1,<0.10", + "databricks-sdk~=0.71", + "databricks-labs-lsql>=0.5,<=0.16", + "sqlalchemy>=1.4,<3.0", ] [project.optional-dependencies] diff --git a/src/databricks/labs/dqx/config.py b/src/databricks/labs/dqx/config.py index f18004cd3..8c3fcee2d 100644 --- a/src/databricks/labs/dqx/config.py +++ b/src/databricks/labs/dqx/config.py @@ -141,6 +141,9 @@ class WorkspaceConfig: # whether to use serverless clusters for the jobs, only used during workspace installation serverless_clusters: bool = True + upload_dependencies: bool = ( + False # whether to upload dependencies to the workspace during installation to enable DQX in restricted (no-internet) environments + ) extra_params: ExtraParams | None = None # extra parameters to pass to the jobs, e.g. result_column_names # cluster configuration for the jobs (applicable for non-serverless clusters only) diff --git a/src/databricks/labs/dqx/engine.py b/src/databricks/labs/dqx/engine.py index f4b8c7ed7..61165b3cd 100644 --- a/src/databricks/labs/dqx/engine.py +++ b/src/databricks/labs/dqx/engine.py @@ -1168,6 +1168,7 @@ def get_streaming_metrics_listener( ) return StreamingMetricsListener(metrics_config, metrics_observation, self.spark, target_query_id) + @telemetry_logger("engine", "apply_checks_for_run_config") def _apply_checks_for_run_config(self, run_config: RunConfig) -> None: """ Applies checks based on a given RunConfig. diff --git a/src/databricks/labs/dqx/installer/config_provider.py b/src/databricks/labs/dqx/installer/config_provider.py index b70862d8d..9d1ad40cc 100644 --- a/src/databricks/labs/dqx/installer/config_provider.py +++ b/src/databricks/labs/dqx/installer/config_provider.py @@ -98,6 +98,9 @@ def prompt_new_installation(self, install_folder: str | None = None) -> Workspac warehouse_id = self._warehouse_configurator.create() + # Ask if the workspace blocks Internet access to determine if dependencies should be uploaded + upload_dependencies = self._prompts.confirm("Does the given workspace block Internet access?") + return WorkspaceConfig( log_level=log_level, run_configs=[ @@ -114,6 +117,7 @@ def prompt_new_installation(self, install_folder: str | None = None) -> Workspac ) ], serverless_clusters=serverless_clusters, + upload_dependencies=upload_dependencies, profiler_spark_conf=profiler_spark_conf, profiler_override_clusters=profiler_override_clusters, quality_checker_spark_conf=quality_checker_spark_conf, diff --git a/src/databricks/labs/dqx/installer/workflow_installer.py b/src/databricks/labs/dqx/installer/workflow_installer.py index 15578a966..6c0bd4959 100644 --- a/src/databricks/labs/dqx/installer/workflow_installer.py +++ b/src/databricks/labs/dqx/installer/workflow_installer.py @@ -1,5 +1,6 @@ from __future__ import annotations +import importlib.metadata import logging import os.path import re @@ -487,9 +488,82 @@ def _library_dep_order(library: str): case _: return 2 + @staticmethod + def _extract_dependency_prefix(requirement: str) -> str | None: + """ + Extract package name prefix from a requirement string. + + Args: + requirement: Requirement string (e.g., "databricks-sdk>=0.71") + + Returns: + Package name prefix with underscores, or None if invalid. + """ + match = re.match(r'^([a-zA-Z0-9-]+)', requirement) + if match: + pkg_name = match.group(1) + return pkg_name.replace('-', '_') + return None + + @staticmethod + def _get_fallback_dependencies() -> list[str]: + """ + Get fallback dependency prefixes when metadata is unavailable. The list should match the dependency list + from the pyproject.toml file in the project root. + + Returns: + List of core dependency prefixes. + """ + return [ + "databricks_labs_blueprint", + "databricks_sdk", + "databricks_labs_lsql", + "sqlalchemy", + ] + + @staticmethod + def _get_dependency_prefixes() -> list[str]: + """ + Dynamically retrieve dependency prefixes from package metadata. + + Includes both core and optional (extras) dependencies to ensure workflows + have access to all required packages. + + Returns: + List of dependency package name prefixes for wheel files. + """ + try: + requires = importlib.metadata.requires('databricks-labs-dqx') + except importlib.metadata.PackageNotFoundError: + logger.warning("databricks-labs-dqx package metadata not found, using fallback dependencies") + return WorkflowDeployment._get_fallback_dependencies() + + if not requires: + logger.warning("No dependencies found in package metadata") + return [] + + prefixes = [] + for req in requires: + prefix = WorkflowDeployment._extract_dependency_prefix(req) + if prefix: + prefixes.append(prefix) + + # Remove duplicates while preserving order + unique_prefixes = list(dict.fromkeys(prefixes)) + logger.info(f"Discovered {len(unique_prefixes)} dependencies from package metadata (including extras)") + return unique_prefixes + def _upload_wheel(self): wheel_paths = [] with self._wheels: + # Upload dependencies if workspace blocks Internet access + if self._config.upload_dependencies: + logger.info("Uploading dependencies to workspace...") + dependency_prefixes = self._get_dependency_prefixes() + # TODO the _build_wheel in the upload wheel dependencies method + # currently does not handle installation of extras, therefore they are not uploaded + for whl in self._wheels.upload_wheel_dependencies(dependency_prefixes): + wheel_paths.append(whl) wheel_paths.sort(key=WorkflowDeployment._library_dep_order) wheel_paths.append(self._wheels.upload_to_wsfs()) wheel_paths = [f"/Workspace{wheel}" for wheel in wheel_paths] diff --git a/tests/e2e/test_run_demos.py b/tests/e2e/test_run_demos.py index 5d38c3ab8..713e891ba 100644 --- a/tests/e2e/test_run_demos.py +++ b/tests/e2e/test_run_demos.py @@ -142,9 +142,9 @@ def test_run_dqx_demo_tool(installation_ctx, make_schema, make_notebook, make_jo schema = make_schema(catalog_name=catalog).name installation_ctx.replace( extend_prompts={ - r'Provide location for the input data .*': '/databricks-datasets/delta-sharing/samples/nyctaxi_2019', - r'Provide output table .*': f'{catalog}.{schema}.output_table', - r'Provide quarantined table .*': f'{catalog}.{schema}.quarantine_table', + r"Provide location for the input data .*": "/databricks-datasets/delta-sharing/samples/nyctaxi_2019", + r"Provide output table .*": f"{catalog}.{schema}.output_table", + r"Provide quarantined table .*": f"{catalog}.{schema}.quarantine_table", }, ) installation_ctx.workspace_installer.run(installation_ctx.config) diff --git a/tests/integration/test_ai_rules_generator.py b/tests/integration/test_ai_rules_generator.py index 2acc139b6..258d4b67a 100644 --- a/tests/integration/test_ai_rules_generator.py +++ b/tests/integration/test_ai_rules_generator.py @@ -9,25 +9,11 @@ # Sample user input with specific requirements to avoid flakiness in tests USER_INPUT = """ -Username should not start with 's' and should not contain more than 20 letters if user id is provided. Error message must be: "Username should not start with 's' and should not contain more than 20 letters if user id is provided" -Apply validation when user_id is not null. Use SQL expression for this check with the following expression: NOT (username LIKE 's%') AND LENGTH(username) <= 20. -Users at age 18 or above must have a valid email address. -Age should be between 0 and 120. +Users at age 18 or above must have a valid email address checked using regex. +Age should be between 0 and 120. Output the rules in the given order. """ EXPECTED_CHECKS = [ - { - "check": { - "arguments": { - "columns": ["username"], - "expression": "NOT (username LIKE 's%') AND LENGTH(username) <= 20", - "msg": "Username should not start with 's' and should not contain more than 20 letters if user id is provided", - }, - "function": "sql_expression", - }, - "criticality": "error", - "filter": "user_id IS NOT NULL", - }, { "check": { "arguments": {"column": "email", "regex": "^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$"}, diff --git a/tests/integration/test_installation.py b/tests/integration/test_installation.py index 650963ba8..00bc1bf2b 100644 --- a/tests/integration/test_installation.py +++ b/tests/integration/test_installation.py @@ -42,9 +42,9 @@ def factory( prompts = MockPrompts( { - r'Provide location for the input data .*': '/', - r'Provide output table .*': 'main.dqx_test.output_table', - r'Do you want to uninstall DQX .*': 'yes', + r"Provide location for the input data .*": "/", + r"Provide output table .*": "main.dqx_test.output_table", + r"Do you want to uninstall DQX .*": "yes", r".*PRO or SERVERLESS SQL warehouse.*": "1", r".*": "", } diff --git a/tests/integration/test_upload_dependencies.py b/tests/integration/test_upload_dependencies.py new file mode 100644 index 000000000..fe5999708 --- /dev/null +++ b/tests/integration/test_upload_dependencies.py @@ -0,0 +1,325 @@ +import logging +import importlib.metadata +from datetime import timedelta +from unittest.mock import patch, create_autospec +import pytest + +from databricks.labs.blueprint.installation import Installation +from databricks.labs.blueprint.installer import InstallState +from databricks.labs.blueprint.tui import MockPrompts +from databricks.labs.blueprint.wheels import ProductInfo, WheelsV2 +from databricks.labs.dqx.config import WorkspaceConfig, InputConfig, OutputConfig, ProfilerConfig +from databricks.labs.dqx.installer.install import WorkspaceInstaller +from databricks.labs.dqx.installer.workflow_installer import WorkflowDeployment, DeployedWorkflows +from databricks.labs.dqx.installer.workflow_task import Task +from databricks.sdk.service.jobs import RunResultState + +logger = logging.getLogger(__name__) + + +@pytest.fixture +def installation_with_upload_deps(ws, make_random): + """Create an installation with upload_dependencies=True.""" + cleanup = [] + + def factory(): + product_info = ProductInfo.for_testing(WorkspaceConfig) + + prompts = MockPrompts( + { + r"Provide location for the input data .*": "main.dqx_test.input_table", + r"Provide output table .*": "main.dqx_test.output_table", + r"Do you want to uninstall DQX .*": "yes", + r".*PRO or SERVERLESS SQL warehouse.*": "1", + r"Does the given workspace block Internet access\?": "yes", # Enable upload_dependencies + r".*": "", + } + ) + + installation = Installation(ws, product_info.product_name()) + installer = WorkspaceInstaller(ws).replace( + installation=installation, + product_info=product_info, + prompts=prompts, + ) + + workspace_config = installer.configure() + + # Verify upload_dependencies is set to True + assert workspace_config.upload_dependencies is True, "upload_dependencies should be True" + + installation.save(workspace_config) + cleanup.append(installation) + + return installation, workspace_config + + yield factory + + for pending in cleanup: + try: + pending.remove() + except Exception as e: + logger.warning(f"Failed to cleanup installation: {e}") + + +def test_installation_with_upload_dependencies(ws, installation_with_upload_deps): + """Test that installation correctly sets upload_dependencies flag.""" + installation, config = installation_with_upload_deps() + + # Verify configuration + assert config.upload_dependencies is True + + # Verify configuration is saved correctly + loaded_config = installation.load(WorkspaceConfig) + assert loaded_config.upload_dependencies is True + + +def test_workflow_deployment_uploads_dependencies(ws, installation_with_upload_deps, make_random): + """Test that workflow deployment uploads dependencies when upload_dependencies is True.""" + installation, config = installation_with_upload_deps() + + product_info = ProductInfo.for_testing(WorkspaceConfig) + # Mock install state since we're just testing wheel upload behavior + install_state = create_autospec(InstallState) + install_state.jobs = {} # Empty dict for jobs + install_state.save.return_value = None # Mock save to avoid NotInstalled error + # Use the installation from our fixture to create WheelsV2 instead of product_info.wheels(ws) + wheels = WheelsV2(installation, product_info) + + # Create a simple task for testing + tasks = [ + Task( + workflow="test_workflow", + name="test_task", + job_cluster="default", + doc="Test task", + fn=lambda: None, + ) + ] + + deployment = WorkflowDeployment( + config=config, + installation=installation, + install_state=install_state, + ws=ws, + wheels=wheels, + product_info=product_info, + tasks=tasks, + ) + + # Mock the wheels methods and job creation to test upload behavior + with ( + patch.object(InstallState, 'from_installation', return_value=install_state), + patch.object(wheels, 'upload_wheel_dependencies') as mock_upload_deps, + patch.object(wheels, 'upload_to_wsfs') as mock_upload_main, + patch.object(ws.jobs, 'create') as mock_create_job, + ): + mock_upload_deps.return_value = [ + "/foo/bar/databricks_sdk-0.71.0-py3-none-any.whl", + "/foo/bar/databricks_labs_lsql-0.5.0-py3-none-any.whl", + ] + mock_upload_main.return_value = "/foo/bar/databricks_labs_dqx-0.1.0-py3-none-any.whl" + mock_create_job.return_value = type('obj', (object,), {'job_id': 123})() + + # Call public method create_jobs() which internally calls _upload_wheel() + deployment.create_jobs() + + # Verify upload_wheel_dependencies was called + mock_upload_deps.assert_called_once() + + # Verify the dependency prefixes passed to upload_wheel_dependencies + call_args = mock_upload_deps.call_args[0][0] + assert isinstance(call_args, list) + assert len(call_args) > 0 + + # Verify core dependencies are included + assert any("databricks_sdk" in dep for dep in call_args) + assert any("databricks_labs_lsql" in dep for dep in call_args) + + # Verify job was created with the uploaded wheels + mock_create_job.assert_called_once() + job_settings = mock_create_job.call_args[1] + assert 'tasks' in job_settings + + +def test_dependency_discovery_includes_extras(ws, installation_with_upload_deps): + """Test that dependency discovery includes extras dependencies when creating jobs.""" + installation, config = installation_with_upload_deps() + + product_info = ProductInfo.for_testing(WorkspaceConfig) + # Mock install state since we're just testing wheel upload behavior + install_state = create_autospec(InstallState) + install_state.jobs = {} # Empty dict for jobs + install_state.save.return_value = None # Mock save to avoid NotInstalled error + # Use the installation from our fixture to create WheelsV2 instead of product_info.wheels(ws) + wheels = WheelsV2(installation, product_info) + + tasks = [ + Task( + workflow="test_workflow", + name="test_task", + job_cluster="default", + doc="Test task", + fn=lambda: None, + ) + ] + + deployment = WorkflowDeployment( + config=config, + installation=installation, + install_state=install_state, + ws=ws, + wheels=wheels, + product_info=product_info, + tasks=tasks, + ) + + # Mock package metadata to include extras dependencies + mock_requires = [ + "databricks-labs-blueprint>=0.9.1,<0.10", + "databricks-sdk~=0.71", + "databricks-labs-lsql>=0.5,<=0.16", + "sqlalchemy>=1.4,<3.0", + "pydantic>=2.0; extra == 'pii'", # extras dependency + "openai>=1.0; extra == 'llm'", # extras dependency + ] + + with ( + patch.object(InstallState, 'from_installation', return_value=install_state), + patch.object(importlib.metadata, 'requires', return_value=mock_requires), + patch.object(wheels, 'upload_wheel_dependencies') as mock_upload_deps, + patch.object(wheels, 'upload_to_wsfs') as mock_upload_main, + patch.object(ws.jobs, 'create') as mock_create_job, + ): + mock_upload_deps.return_value = ["/foo/bar/dep.whl"] + mock_upload_main.return_value = "/foo/bar/dqx.whl" + mock_create_job.return_value = type('obj', (object,), {'job_id': 123})() + + # Call public method that triggers dependency discovery + deployment.create_jobs() + + # Verify upload_wheel_dependencies was called with all dependencies including extras + mock_upload_deps.assert_called_once() + uploaded_prefixes = mock_upload_deps.call_args[0][0] + + # Verify all dependencies are included (core + extras) + assert len(uploaded_prefixes) == 6 + assert "databricks_labs_blueprint" in uploaded_prefixes + assert "databricks_sdk" in uploaded_prefixes + assert "databricks_labs_lsql" in uploaded_prefixes + assert "sqlalchemy" in uploaded_prefixes + assert "pydantic" in uploaded_prefixes # extras should be included + assert "openai" in uploaded_prefixes # extras should be included + + +def test_config_upload_dependencies_persists(ws, installation_with_upload_deps): + """Test that upload_dependencies configuration persists through save/load cycle.""" + installation, original_config = installation_with_upload_deps() + + # Verify original config + assert original_config.upload_dependencies is True + + # Load config from installation + loaded_config = installation.load(WorkspaceConfig) + + # Verify loaded config has the same value + assert loaded_config.upload_dependencies is True + + # Verify it's properly serialized in as_dict() + config_dict = loaded_config.as_dict() + assert "upload_dependencies" in config_dict + assert config_dict["upload_dependencies"] is True + + +def test_end_to_end_installation_and_workflow_with_upload_dependencies(ws, make_schema, make_table, make_random): + """ + End-to-end integration test: Install DQX with upload_dependencies=True and run a workflow. + This verifies that dependencies are properly uploaded and workflows can execute successfully. + """ + product_info = ProductInfo.for_testing(WorkspaceConfig) + + # Prepare test data + catalog_name = "main" + schema = make_schema(catalog_name=catalog_name) + input_table = make_table( + catalog_name=catalog_name, + schema_name=schema.name, + ctas="SELECT * FROM VALUES (1, 'a'), (2, 'b'), (3, NULL) AS data(id, name)", + ) + + # Create prompts with upload_dependencies enabled + prompts = MockPrompts( + { + r"Provide location for the input data .*": input_table.full_name, + r"Provide output table .*": f"{catalog_name}.{schema.name}.{make_random(10).lower()}", + r"Do you want to uninstall DQX .*": "yes", + r".*PRO or SERVERLESS SQL warehouse.*": "1", + r"Does the given workspace block Internet access\?": "yes", # Enable upload_dependencies + r".*": "", + } + ) + + installation = Installation(ws, product_info.product_name()) + installer = WorkspaceInstaller(ws).replace( + installation=installation, + product_info=product_info, + prompts=prompts, + ) + + try: + # Configure and run installation (creates jobs and uploads dependencies) + workspace_config = _configure_test_workspace(installer, input_table, catalog_name, schema.name, make_random) + installation.save(workspace_config) + + # Complete the installation by running the installer + # This creates jobs, uploads dependencies, etc. + installer.run(default_config=workspace_config) + + # Run and verify profiler workflow + run_id = _run_and_verify_workflow(ws, installation) + logger.info(f"✅ End-to-end test passed: Workflow {run_id} completed with upload_dependencies=True") + + finally: + try: + installation.remove() + except Exception as e: + logger.warning(f"Failed to cleanup installation: {e}") + + +def _configure_test_workspace(installer, input_table, catalog_name, schema_name, make_random): + """Helper to configure workspace with test data.""" + workspace_config = installer.configure() + assert workspace_config.upload_dependencies is True, "upload_dependencies should be True" + + run_config = workspace_config.get_run_config() + run_config.input_config = InputConfig(location=input_table.full_name, options={"versionAsOf": "0"}) + output_table = f"{catalog_name}.{schema_name}.{make_random(10).lower()}" + run_config.output_config = OutputConfig(location=output_table) + run_config.profiler_config = ProfilerConfig(sample_fraction=1.0, sample_seed=100) + + return workspace_config + + +def _run_and_verify_workflow(ws, installation): + """Helper to run and verify profiler workflow execution.""" + install_state = InstallState.from_installation(installation) + assert len(install_state.jobs) > 0, "Jobs should be created" + + deployed_workflows = DeployedWorkflows(ws, install_state) + run_id = deployed_workflows.run_workflow("profiler", run_config_name="default", max_wait=timedelta(minutes=15)) + + assert run_id is not None, "Workflow should return a run_id" + + # Get the run details using get_run instead of list_runs + job_run = ws.jobs.get_run(run_id=run_id) + assert job_run is not None, "Job run should exist" + + run_state = job_run.state + assert run_state is not None, "Job run should have a state" + + assert run_state.result_state in [ + RunResultState.SUCCESS, + RunResultState.CANCELED, + ], f"Job should complete successfully, got: {run_state.result_state}" + + return run_id diff --git a/tests/perf/generate_md_report.py b/tests/perf/generate_md_report.py index a7abaf6cd..8d52eee64 100644 --- a/tests/perf/generate_md_report.py +++ b/tests/perf/generate_md_report.py @@ -29,10 +29,15 @@ lines.append("# Performance Benchmarks Report\n") -lines.append("## Specification") +lines.append("## Specification\n") lines.append("* 100 million rows are used for each test.") -lines.append("* Tests are run sequentially to reduce variability.") -lines.append("* Benchmarks are created using Databricks Serverless cluster.\n") +lines.append( + "* DQX rules are executed in parallel and in distributed manner using Spark. To minimize test variability, each type of check is executed sequentially in this benchmark. Specific tests, such as `test_benchmark_apply_checks_all_dataset_checks` and `test_benchmark_apply_checks_all_row_checks`, execute all types of checks in parallel simultaneously." +) +lines.append("* Benchmarks are created using Databricks Serverless cluster.") +lines.append( + "* The provided benchmarks are indicative. You should always consider benchmarking results in the context of your own data and environment.\n" +) lines.append("## Results") diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index af5c429d5..770ca22fb 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -1,11 +1,19 @@ import pytest -from databricks.labs.dqx.config import WorkspaceConfig, RunConfig, InstallationChecksStorageConfig from databricks.labs.dqx.config import ( + WorkspaceConfig, + RunConfig, + InstallationChecksStorageConfig, FileChecksStorageConfig, LakebaseChecksStorageConfig, WorkspaceFileChecksStorageConfig, TableChecksStorageConfig, VolumeFileChecksStorageConfig, + InputConfig, + OutputConfig, + ProfilerConfig, + LLMModelConfig, + LLMConfig, + ExtraParams, ) from databricks.labs.dqx.errors import InvalidConfigError, InvalidParameterError @@ -140,3 +148,303 @@ def test_get_run_config_when_no_run_configs(): def test_post_init_validation(config_class, location, expected_exception, expected_message): with pytest.raises(expected_exception, match=expected_message): config_class(location=location) + + +# Test InputConfig +def test_input_config_defaults(): + config = InputConfig(location="table_name") + assert config.location == "table_name" + assert config.format == "delta" + assert config.is_streaming is False + assert config.schema is None + assert not config.options + + +def test_input_config_with_options(): + config = InputConfig( + location="s3://bucket/path", + format="parquet", + is_streaming=True, + schema="my_schema", + options={"header": "true", "delimiter": ","}, + ) + assert config.location == "s3://bucket/path" + assert config.format == "parquet" + assert config.is_streaming is True + assert config.schema == "my_schema" + assert config.options == {"header": "true", "delimiter": ","} + + +# Test OutputConfig +def test_output_config_defaults(): + config = OutputConfig(location="output_table") + assert config.location == "output_table" + assert config.format == "delta" + assert config.mode == "append" + assert not config.options + assert not config.trigger + + +def test_output_config_trigger_string_to_bool_conversion(): + config = OutputConfig( + location="output_table", + trigger={"once": "true", "availableNow": "True", "continuous": "false", "processingTime": "10 seconds"}, + ) + assert config.trigger["once"] is True + assert config.trigger["availableNow"] is True + assert config.trigger["continuous"] is False + assert config.trigger["processingTime"] == "10 seconds" + + +def test_output_config_trigger_with_actual_booleans(): + config = OutputConfig(location="output_table", trigger={"once": True, "continuous": False}) + assert config.trigger["once"] is True + assert config.trigger["continuous"] is False + + +# Test ProfilerConfig +def test_profiler_config_defaults(): + config = ProfilerConfig() + assert config.summary_stats_file == "profile_summary_stats.yml" + assert config.sample_fraction == 0.3 + assert config.sample_seed is None + assert config.limit == 1000 + assert config.filter is None + + +def test_profiler_config_custom_values(): + config = ProfilerConfig( + summary_stats_file="custom_stats.yml", + sample_fraction=0.5, + sample_seed=42, + limit=5000, + filter="col1 > 0", + ) + assert config.summary_stats_file == "custom_stats.yml" + assert config.sample_fraction == 0.5 + assert config.sample_seed == 42 + assert config.limit == 5000 + assert config.filter == "col1 > 0" + + +# Test LLMModelConfig +def test_llm_model_config_defaults(): + config = LLMModelConfig() + assert config.model_name == "databricks/databricks-claude-sonnet-4-5" + assert config.api_key == "" + assert config.api_base == "" + + +def test_llm_model_config_custom_values(): + config = LLMModelConfig( + model_name="custom-model", api_key="secret_scope/secret_key", api_base="https://api.example.com" + ) + assert config.model_name == "custom-model" + assert config.api_key == "secret_scope/secret_key" + assert config.api_base == "https://api.example.com" + + +# Test LLMConfig +def test_llm_config_defaults(): + config = LLMConfig() + assert isinstance(config.model, LLMModelConfig) + assert config.model.model_name == "databricks/databricks-claude-sonnet-4-5" + + +def test_llm_config_with_custom_model(): + model = LLMModelConfig(model_name="custom-model") + config = LLMConfig(model=model) + assert config.model.model_name == "custom-model" + + +# Test ExtraParams +def test_extra_params_defaults(): + config = ExtraParams() + assert not config.result_column_names + assert not config.user_metadata + assert config.run_time_overwrite is None + assert config.run_id_overwrite is None + + +def test_extra_params_custom_values(): + config = ExtraParams( + result_column_names={"col1": "renamed_col1"}, + user_metadata={"key": "value"}, + run_time_overwrite="2025-01-01", + run_id_overwrite="custom_run_id", + ) + assert config.result_column_names == {"col1": "renamed_col1"} + assert config.user_metadata == {"key": "value"} + assert config.run_time_overwrite == "2025-01-01" + assert config.run_id_overwrite == "custom_run_id" + + +# Test WorkspaceConfig.as_dict() +def test_workspace_config_as_dict(): + config = WorkspaceConfig( + run_configs=[DEFAULT_RUN_CONFIG], + log_level="DEBUG", + serverless_clusters=False, + ) + result = config.as_dict() + assert isinstance(result, dict) + assert result["log_level"] == "DEBUG" + assert result["serverless_clusters"] is False + assert len(result["run_configs"]) == 1 + # Verify the method properly converts the config to a dictionary + assert "profiler_max_parallelism" in result + assert "quality_checker_max_parallelism" in result + + +# Test LakebaseChecksStorageConfig validation and properties +def test_lakebase_config_missing_instance_name(): + with pytest.raises(InvalidParameterError, match="Instance name must not be empty or None"): + LakebaseChecksStorageConfig(location="db.schema.table", instance_name=None, user="test_user") + + +def test_lakebase_config_missing_user(): + with pytest.raises(InvalidParameterError, match="User must not be empty or None"): + LakebaseChecksStorageConfig(location="db.schema.table", instance_name="instance", user=None) + + +def test_lakebase_config_invalid_location_format(): + with pytest.raises(InvalidConfigError, match="Invalid Lakebase table name.*Must be in the format"): + LakebaseChecksStorageConfig(location="invalid_table", instance_name="instance", user="test_user") + + +def test_lakebase_config_invalid_mode(): + with pytest.raises(InvalidConfigError, match="Invalid mode.*Must be 'append' or 'overwrite'"): + LakebaseChecksStorageConfig( + location="db.schema.table", instance_name="instance", user="test_user", mode="invalid" + ) + + +def test_lakebase_config_properties(): + config = LakebaseChecksStorageConfig( + location="my_db.my_schema.my_table", instance_name="instance", user="test_user" + ) + assert config.database_name == "my_db" + assert config.schema_name == "my_schema" + assert config.table_name == "my_table" + + +def test_lakebase_config_properties_cached(): + config = LakebaseChecksStorageConfig(location="db1.sch1.tbl1", instance_name="instance", user="test_user") + # Access properties multiple times to ensure caching works + assert config.database_name == "db1" + assert config.database_name == "db1" + assert config.schema_name == "sch1" + assert config.schema_name == "sch1" + assert config.table_name == "tbl1" + assert config.table_name == "tbl1" + + +def test_lakebase_config_valid_modes(): + config_append = LakebaseChecksStorageConfig( + location="db.schema.table", instance_name="instance", user="test_user", mode="append" + ) + assert config_append.mode == "append" + + config_overwrite = LakebaseChecksStorageConfig( + location="db.schema.table", instance_name="instance", user="test_user", mode="overwrite" + ) + assert config_overwrite.mode == "overwrite" + + +# Test VolumeFileChecksStorageConfig valid paths +def test_volume_file_config_valid_yml_path(): + config = VolumeFileChecksStorageConfig(location="/Volumes/main/demo/files/checks.yml") + assert config.location == "/Volumes/main/demo/files/checks.yml" + + +def test_volume_file_config_valid_yaml_path(): + config = VolumeFileChecksStorageConfig(location="/Volumes/catalog/schema/volume/subfolder/checks.yaml") + assert config.location == "/Volumes/catalog/schema/volume/subfolder/checks.yaml" + + +def test_volume_file_config_valid_json_path(): + config = VolumeFileChecksStorageConfig(location="/Volumes/cat/sch/vol/checks.json") + assert config.location == "/Volumes/cat/sch/vol/checks.json" + + +# Test TableChecksStorageConfig with defaults +def test_table_config_defaults(): + config = TableChecksStorageConfig(location="catalog.schema.table") + assert config.location == "catalog.schema.table" + assert config.run_config_name == "default" + assert config.mode == "overwrite" + + +def test_table_config_custom_values(): + config = TableChecksStorageConfig(location="my_table", run_config_name="custom_run", mode="append") + assert config.location == "my_table" + assert config.run_config_name == "custom_run" + assert config.mode == "append" + + +# Test FileChecksStorageConfig valid path +def test_file_config_valid_path(): + config = FileChecksStorageConfig(location="/path/to/checks.yml") + assert config.location == "/path/to/checks.yml" + + +# Test WorkspaceFileChecksStorageConfig valid path +def test_workspace_file_config_valid_path(): + config = WorkspaceFileChecksStorageConfig(location="/Workspace/Users/user@example.com/checks.yml") + assert config.location == "/Workspace/Users/user@example.com/checks.yml" + + +# Test InstallationChecksStorageConfig defaults +def test_installation_config_defaults(): + config = InstallationChecksStorageConfig(location="installation") + assert config.location == "installation" + assert config.run_config_name == "default" + assert config.product_name == "dqx" + assert config.assume_user is True + assert config.install_folder is None + assert config.overwrite_location is False + + +def test_installation_config_custom_values(): + config = InstallationChecksStorageConfig( + location="custom_location", + run_config_name="custom_run", + product_name="custom_product", + assume_user=False, + install_folder="/custom/folder", + overwrite_location=True, + ) + assert config.location == "custom_location" + assert config.run_config_name == "custom_run" + assert config.product_name == "custom_product" + assert config.assume_user is False + assert config.install_folder == "/custom/folder" + assert config.overwrite_location is True + + +# Test RunConfig +def test_run_config_defaults(): + config = RunConfig() + assert config.name == "default" + assert config.input_config is None + assert config.output_config is None + assert config.quarantine_config is None + assert config.metrics_config is None + assert isinstance(config.profiler_config, ProfilerConfig) + assert config.checks_user_requirements is None + assert config.checks_location == "checks.yml" + assert config.warehouse_id is None + assert not config.reference_tables + assert not config.custom_check_functions + assert config.lakebase_instance_name is None + assert config.lakebase_user is None + assert config.lakebase_port is None + + +def test_run_config_with_input_output(): + input_cfg = InputConfig(location="input_table") + output_cfg = OutputConfig(location="output_table") + config = RunConfig(name="test_run", input_config=input_cfg, output_config=output_cfg) + assert config.name == "test_run" + assert config.input_config == input_cfg + assert config.output_config == output_cfg