OBSL-1013: ship table discovery in soda-core (soda data-source discover)

soda-core/src/soda_core/cli/cli.py

@@ -17,6 +17,7 @@
 )
 from soda_core.cli.handlers.data_source import (
     handle_create_data_source,
+    handle_discover_data_source,
     handle_test_data_source,
 )
 from soda_core.cli.handlers.request import (
@@ -425,6 +426,7 @@ def _setup_data_source_resource(resource_parsers) -> None:
 
     _setup_data_source_create_command(data_source_subparsers)
     _setup_data_source_test_command(data_source_subparsers)
+    _setup_data_source_discover_command(data_source_subparsers)
 
 
 def _setup_data_source_create_command(data_source_parsers) -> None:
@@ -488,6 +490,39 @@ def handle(args):
     test_parser.set_defaults(handler_func=handle)
 
 
+def _setup_data_source_discover_command(data_source_parsers) -> None:
+    discover_parser = data_source_parsers.add_parser("discover", help="Discover datasets in a data source")
+    discover_parser.add_argument("-ds", "--data-source", type=str, help="The data source configuration file.")
+    discover_parser.add_argument(
+        "--include", type=str, nargs="*", help="Dataset name patterns to include (SQL %% wildcard)."
+    )
+    discover_parser.add_argument(
+        "--exclude", type=str, nargs="*", help="Dataset name patterns to exclude (SQL %% wildcard)."
+    )
+    discover_parser.add_argument("--scan-definition-name", type=str, help="Override the scan definition name.")
+    discover_parser.add_argument("-sc", "--soda-cloud", type=str, help=CLOUD_CONFIG_PATH_HELP)
+    discover_parser.add_argument(
+        "-v",
+        "--verbose",
+        const=True,
+        action="store_const",
+        default=False,
+        help="Show more detailed logs on the console.",
+    )
+
+    def handle(args):
+        exit_code = handle_discover_data_source(
+            args.data_source,
+            args.include,
+            args.exclude,
+            args.scan_definition_name,
+            soda_cloud_file_path=args.soda_cloud,
+        )
+        exit_with_code(exit_code)
+
+    discover_parser.set_defaults(handler_func=handle)
+
+
 def _setup_soda_cloud_resource(resource_parsers) -> None:
     soda_cloud_parser = resource_parsers.add_parser("cloud", help="Soda Cloud commands")
     soda_cloud_subparsers = soda_cloud_parser.add_subparsers(dest="command", help="Soda Cloud commands")

soda-core/src/soda_core/cli/handlers/data_source.py

@@ -126,3 +126,67 @@ def build_test_connection_log_uploader(
         dataset="",
     )
     return Logs(gatherer=logs_queue)
+
+
+def handle_discover_data_source(
+    data_source_file_path: str,
+    include: Optional[list[str]] = None,
+    exclude: Optional[list[str]] = None,
+    scan_definition_name: Optional[str] = None,
+    soda_cloud_file_path: Optional[str] = None,
+) -> ExitCode:
+    from soda_core.common.data_source_impl import DataSourceImpl
+    from soda_core.discovery.discovery_payload import (
+        build_discovery_payload,
+        send_discovery_results,
+    )
+    from soda_core.discovery.discovery_run import DiscoveryRun
+
+    soda_logger.info(f"Discovering datasets for data source configuration file {data_source_file_path}")
+    data_source_impl: Optional[DataSourceImpl] = DataSourceImpl.from_yaml_source(
+        DataSourceYamlSource.from_file_path(data_source_file_path)
+    )
+    if data_source_impl is None:
+        soda_logger.error(f"{Emoticons.POLICE_CAR_LIGHT} Data source could not be created. See logs above (or -v).")
+        return ExitCode.LOG_ERRORS
+
+    if not soda_cloud_file_path:
+        soda_logger.error(f"{Emoticons.POLICE_CAR_LIGHT} Discovery requires a Soda Cloud configuration (-sc).")
+        return ExitCode.LOG_ERRORS
+    soda_cloud: Optional[SodaCloud] = SodaCloud.from_yaml_source(
+        SodaCloudYamlSource.from_file_path(soda_cloud_file_path),
+        provided_variable_values=None,
+    )
+    if soda_cloud is None:
+        soda_logger.error(f"{Emoticons.POLICE_CAR_LIGHT} Soda Cloud configuration could not be parsed.")
+        return ExitCode.LOG_ERRORS
+
+    try:
+        # from_yaml_source only parses YAML; the handler owns the connection lifecycle.
+        data_source_impl.open_connection()
+        # Empty prefixes: discover everything visible to the connection (v3 behaviour).
+        dqns: list[str] = DiscoveryRun.execute(
+            data_source_impl=data_source_impl,
+            prefixes=[],
+            include=include,
+            exclude=exclude,
+        )
+    except Exception as exc:
+        soda_logger.exception(f"Discovery query failed: {exc}")
+        return ExitCode.LOG_ERRORS
+    finally:
+        data_source_impl.close_connection()
+
+    resolved_scan_definition_name: str = scan_definition_name or f"{data_source_impl.name}_schema_discovery_scan"
+    payload: dict = build_discovery_payload(
+        dqns=dqns,
+        data_source_name=data_source_impl.name,
+        scan_definition_name=resolved_scan_definition_name,
+    )
+    response = send_discovery_results(soda_cloud, payload)
+    if response is None or not response.ok:
+        soda_logger.error(f"{Emoticons.POLICE_CAR_LIGHT} Discovery results were not accepted by Soda Cloud.")
+        return ExitCode.RESULTS_NOT_SENT_TO_CLOUD
+
+    soda_logger.info(f"{Emoticons.WHITE_CHECK_MARK} Discovered {len(dqns)} datasets and sent results to Soda Cloud.")
+    return ExitCode.OK

soda-core/src/soda_core/common/data_source_impl.py

@@ -402,6 +402,8 @@ def discover_qualified_objects(
         self,
         prefixes: list[str],
         object_types: Optional[list[TableType]] = None,
+        include_table_name_like_filters: Optional[list[str]] = None,
+        exclude_table_name_like_filters: Optional[list[str]] = None,
     ) -> list[FullyQualifiedObjectName]:
         metadata_tables_query: MetadataTablesQuery = self.create_metadata_tables_query()
         database_name = self.extract_database_from_prefix(prefixes)
@@ -411,6 +413,8 @@ def discover_qualified_objects(
             database_name=database_name,
             schema_name=schema_name,
             types_to_return=object_types,
+            include_table_name_like_filters=include_table_name_like_filters,
+            exclude_table_name_like_filters=exclude_table_name_like_filters,
         )
         return fully_qualified_object_names
 

soda-core/src/soda_core/common/dataset_identifier.py

@@ -32,6 +32,28 @@ def parse(cls, dataset_qualified_name: Optional[str]) -> DatasetIdentifier:
 
         return cls(data_source_name, prefixes, dataset_name)
 
+    @classmethod
+    def from_object(cls, data_source_name, sql_dialect, fully_qualified_object_name) -> "DatasetIdentifier":
+        """Build a dialect-correct DQN from a discovered FullyQualifiedObjectName.
+
+        A prefix component is included only when the dialect has that tier
+        (prefix-index hook not None) and the object carries a value;
+        database precedes schema, as in extract_database_from_prefix.
+        """
+        prefixes: list[str] = []
+        if (
+            sql_dialect.get_database_prefix_index() is not None
+            and fully_qualified_object_name.database_name is not None
+        ):
+            prefixes.append(fully_qualified_object_name.database_name)
+        if sql_dialect.get_schema_prefix_index() is not None and fully_qualified_object_name.schema_name is not None:
+            prefixes.append(fully_qualified_object_name.schema_name)
+        return cls(
+            data_source_name=data_source_name,
+            prefixes=prefixes,
+            dataset_name=fully_qualified_object_name.get_object_name(),
+        )
+
     def to_string(self) -> str:
         return "/".join([self.data_source_name] + self.prefixes + [self.dataset_name])
 

soda-core/src/soda_core/discovery/discovery_payload.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from requests import Response
+from soda_core.common.soda_cloud import SodaCloud
+
+
+def build_discovery_payload(dqns: list[str], data_source_name: str, scan_definition_name: str) -> dict:
+    """DQN-only sodaCoreInsertScanResults body for v4 discovery. Mirrors the non-routing
+    fields v3 emits for BE DTO deserialization safety."""
+    return {
+        "type": "sodaCoreInsertScanResults",
+        "version": "4",
+        "scanType": None,
+        "definitionName": scan_definition_name,
+        "defaultDataSource": data_source_name,
+        "metadata": [{"datasetQualifiedName": dqn} for dqn in dqns],
+        "checks": [],
+        "metrics": [],
+        "profiling": [],
+        "automatedMonitoringChecks": [],
+    }
+
+
+def send_discovery_results(soda_cloud: SodaCloud, payload: dict) -> Optional[Response]:
+    return soda_cloud._execute_command(command_json_dict=payload, request_log_name="send_discovery_results")

soda-core/src/soda_core/discovery/discovery_run.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from soda_core.common.data_source_impl import DataSourceImpl
+from soda_core.common.dataset_identifier import DatasetIdentifier
+
+SODA_TEMP_PREFIX = "__soda_temp"
+
+
+class DiscoveryRun:
+    """Discovers dataset names for a data source and returns their DQNs."""
+
+    @staticmethod
+    def execute(
+        data_source_impl: DataSourceImpl,
+        prefixes: list[str],
+        include: Optional[list[str]] = None,
+        exclude: Optional[list[str]] = None,
+    ) -> list[str]:
+        # include/exclude are pushed down as SQL LIKE filters. The __soda_temp prefix is
+        # filtered in Python instead: in a LIKE pattern its leading underscores would be
+        # single-character wildcards.
+        objects = data_source_impl.discover_qualified_objects(
+            prefixes=prefixes,
+            include_table_name_like_filters=include,
+            exclude_table_name_like_filters=exclude,
+        )
+        objects = [o for o in objects if not o.get_object_name().lower().startswith(SODA_TEMP_PREFIX)]
+        return [
+            DatasetIdentifier.from_object(data_source_impl.name, data_source_impl.sql_dialect, o).to_string()
+            for o in objects
+        ]

soda-tests/tests/integration/test_discovery.py

@@ -0,0 +1,94 @@
+from textwrap import dedent
+
+import pytest
+from helpers.data_source_test_helper import DataSourceTestHelper
+from helpers.mock_soda_cloud import MockResponse, MockSodaCloud
+from helpers.test_fixtures import test_datasource
+from helpers.test_table import TestTableSpecification
+from soda_core.cli.exit_codes import ExitCode
+from soda_core.cli.handlers.data_source import handle_discover_data_source
+from soda_core.common.soda_cloud import SodaCloud
+from soda_core.discovery.discovery_payload import (
+    build_discovery_payload,
+    send_discovery_results,
+)
+from soda_core.discovery.discovery_run import DiscoveryRun
+
+test_table_specification = (
+    TestTableSpecification.builder()
+    .table_purpose("discovery")
+    .column_varchar("id")
+    .column_integer("age")
+    .rows(rows=[("1", 30), ("2", 40)])
+    .build()
+)
+
+
+def test_discovery_posts_dqn_only_v4_payload(data_source_test_helper: DataSourceTestHelper):
+    test_table = data_source_test_helper.ensure_test_table(test_table_specification)
+    data_source_test_helper.enable_soda_cloud_mock(
+        [MockResponse(status_code=200, json_object={"scanId": "discovery_scan"})]
+    )
+
+    # Scope discovery to the test schema for determinism (shared CI DB has many schemas).
+    prefixes = data_source_test_helper._create_dataset_prefix()
+    dqns = DiscoveryRun.execute(data_source_test_helper.data_source_impl, prefixes=prefixes)
+
+    payload = build_discovery_payload(
+        dqns=dqns,
+        data_source_name=data_source_test_helper.data_source_impl.name,
+        scan_definition_name=f"{data_source_test_helper.data_source_impl.name}_schema_discovery_scan",
+    )
+    send_discovery_results(data_source_test_helper.soda_cloud, payload)
+
+    posted = data_source_test_helper.soda_cloud.requests[0].json
+    assert posted["type"] == "sodaCoreInsertScanResults"
+    assert posted["version"] == "4"
+    assert all(set(entry.keys()) == {"datasetQualifiedName"} for entry in posted["metadata"])
+    expected_suffix = test_table.unique_name.lower()
+    assert any(
+        entry["datasetQualifiedName"].lower().endswith(expected_suffix) for entry in posted["metadata"]
+    ), f"{test_table.unique_name} not found in {posted['metadata']}"
+
+
+@pytest.mark.no_snapshot  # Opens its own connection from the YAML file, outside the helper's snapshot machinery.
+@pytest.mark.skipif(
+    test_datasource not in {"postgres"},
+    reason=(
+        "The handler connection lifecycle (open/close around the discovery "
+        "query) is dialect-independent, so one dialect suffices. The handler "
+        "discovers with unscoped prefixes, which doesn't surface the test "
+        "table on some dialects (e.g. databricks/sparkdf shared metastores)."
+    ),
+)
+def test_handle_discover_data_source_opens_connection_and_posts_payload(
+    data_source_test_helper: DataSourceTestHelper, tmp_path, monkeypatch
+):
+    """Regression test for the `soda data-source discover` CLI path:
+    from_yaml_source does not open the connection, so the handler must
+    open/close it around the discovery query.
+    """
+    test_table = data_source_test_helper.ensure_test_table(test_table_specification)
+
+    data_source_file = tmp_path / "data_source.yml"
+    data_source_file.write_text(dedent(data_source_test_helper._create_data_source_yaml_str()).strip())
+    soda_cloud_file = tmp_path / "soda_cloud.yml"
+    soda_cloud_file.write_text("soda_cloud:\n  host: mock.soda.io\n")
+
+    mock_soda_cloud = MockSodaCloud([MockResponse(status_code=200, json_object={"scanId": "discovery_scan"})])
+    monkeypatch.setattr(SodaCloud, "from_yaml_source", classmethod(lambda cls, *args, **kwargs: mock_soda_cloud))
+
+    exit_code = handle_discover_data_source(
+        data_source_file_path=str(data_source_file),
+        include=[test_table.unique_name],
+        soda_cloud_file_path=str(soda_cloud_file),
+    )
+
+    assert exit_code == ExitCode.OK
+    assert len(mock_soda_cloud.requests) == 1
+    posted = mock_soda_cloud.requests[0].json
+    assert posted["type"] == "sodaCoreInsertScanResults"
+    expected_suffix = test_table.unique_name.lower()
+    assert any(
+        entry["datasetQualifiedName"].lower().endswith(expected_suffix) for entry in posted["metadata"]
+    ), f"{test_table.unique_name} not found in {posted['metadata']}"

soda-tests/tests/unit/test_cli_discover.py

@@ -0,0 +1,31 @@
+import sys
+from unittest.mock import patch
+
+import pytest
+from soda_core.cli.cli import create_cli_parser
+from soda_core.cli.exit_codes import ExitCode
+
+
+@patch("soda_core.cli.cli.handle_discover_data_source")
+def test_cli_arg_mapping_for_data_source_discover(mock_handler):
+    mock_handler.return_value = ExitCode.OK
+    sys.argv = [
+        "soda",
+        "data-source",
+        "discover",
+        "-ds",
+        "ds.yaml",
+        "--include",
+        "cust%",
+        "--exclude",
+        "tmp%",
+        "--scan-definition-name",
+        "my_scan",
+        "-sc",
+        "cloud.yaml",
+    ]
+    args = create_cli_parser().parse_args()
+    with pytest.raises(SystemExit) as e:
+        args.handler_func(args)
+    assert e.value.code == ExitCode.OK
+    mock_handler.assert_called_once_with("ds.yaml", ["cust%"], ["tmp%"], "my_scan", soda_cloud_file_path="cloud.yaml")

soda-tests/tests/unit/test_dataset_identifier_from_object.py

@@ -0,0 +1,34 @@
+from soda_core.common.dataset_identifier import DatasetIdentifier
+from soda_core.common.statements.table_types import FullyQualifiedTableName
+
+
+class _FakeDialect:
+    """Stands in for a SqlDialect: only the two prefix-index hooks are used."""
+
+    def __init__(self, database_prefix_index, schema_prefix_index):
+        self._db = database_prefix_index
+        self._schema = schema_prefix_index
+
+    def get_database_prefix_index(self):
+        return self._db
+
+    def get_schema_prefix_index(self):
+        return self._schema
+
+
+def test_from_object_postgres_like_includes_database():
+    obj = FullyQualifiedTableName(database_name="soda", schema_name="public", table_name="customers")
+    di = DatasetIdentifier.from_object("postgres", _FakeDialect(0, 1), obj)
+    assert di.to_string() == "postgres/soda/public/customers"
+
+
+def test_from_object_duckdb_like_drops_catalog():
+    obj = FullyQualifiedTableName(database_name="memory", schema_name="main", table_name="t")
+    di = DatasetIdentifier.from_object("dd", _FakeDialect(None, 0), obj)
+    assert di.to_string() == "dd/main/t"
+
+
+def test_from_object_no_database_value():
+    obj = FullyQualifiedTableName(database_name=None, schema_name="HR", table_name="EMP")
+    di = DatasetIdentifier.from_object("ora", _FakeDialect(0, 1), obj)
+    assert di.to_string() == "ora/HR/EMP"