Merge branch 'master' into feat/elasticsearch-optimization-ext

docker/kafka-setup/kafka-topic-workers.sh

@@ -63,3 +63,4 @@ for ((i=1;i<=$WORKERS;i++)); do
   echo will start $i
   work $i &
 done
+

docs/how/updating-datahub.md

@@ -7,6 +7,7 @@ This file documents any backwards-incompatible changes in DataHub and assists pe
 ### Breaking Changes
 
 - #6742 The metadata file sink's output format no longer contains nested JSON strings for MCP aspects, but instead unpacks the stringified JSON into a real JSON object. The previous sink behavior can be recovered using the `legacy_nested_json_string` option. The file source is backwards compatible and supports both formats.
+- #6901 The `env` and `database_alias` fields have been marked deprecated across all sources. We recommend using `platform_instance` where possible instead.
 
 ### Potential Downtime
 

metadata-ingestion/src/datahub/configuration/common.py

@@ -158,6 +158,10 @@ class IgnorableError(MetaError):
     """An error that can be ignored."""
 
 
+class ConfigurationWarning(Warning):
+    """A configuration warning."""
+
+
 class ConfigurationMechanism(ABC):
     @abstractmethod
     def load_config(self, config_fp: IO) -> dict:

metadata-ingestion/src/datahub/configuration/pydantic_field_deprecation.py

@@ -0,0 +1,20 @@
+import warnings
+from typing import Optional, Type
+
+import pydantic
+
+from datahub.configuration.common import ConfigurationWarning
+
+
+def pydantic_field_deprecated(field: str, message: Optional[str] = None) -> classmethod:
+    if message:
+        output = message
+    else:
+        output = f"{field} is deprecated and will be removed in a future release. Please remove it from your config."
+
+    def _validate_deprecated(cls: Type, values: dict) -> dict:
+        if field in values:
+            warnings.warn(output, ConfigurationWarning, stacklevel=2)
+        return values
+
+    return pydantic.root_validator(pre=True, allow_reuse=True)(_validate_deprecated)

metadata-ingestion/src/datahub/configuration/source_common.py

@@ -4,6 +4,7 @@
 from pydantic.fields import Field
 
 from datahub.configuration.common import ConfigModel, ConfigurationError
+from datahub.configuration.pydantic_field_deprecation import pydantic_field_deprecated
 from datahub.metadata.schema_classes import FabricTypeClass
 
 DEFAULT_ENV = FabricTypeClass.PROD
@@ -39,6 +40,11 @@ class EnvBasedSourceConfigBase(ConfigModel):
         description="The environment that all assets produced by this connector belong to",
     )
 
+    _env_deprecation = pydantic_field_deprecated(
+        "env",
+        "env is deprecated and will be removed in a future release. Please use platform_instance instead.",
+    )
+
     @validator("env")
     def env_must_be_one_of(cls, v: str) -> str:
         if v.upper() not in ALL_ENV_TYPES:

metadata-ingestion/src/datahub/configuration/validate_field_removal.py

@@ -3,6 +3,8 @@
 
 import pydantic
 
+from datahub.configuration.common import ConfigurationWarning
+
 
 def pydantic_removed_field(
     field: str,
@@ -13,7 +15,7 @@ def _validate_field_rename(cls: Type, values: dict) -> dict:
             if print_warning:
                 warnings.warn(
                     f"The {field} was removed, please remove it from your recipe.",
-                    UserWarning,
+                    ConfigurationWarning,
                     stacklevel=2,
                 )
             values.pop(field)

metadata-ingestion/src/datahub/configuration/validate_field_rename.py

@@ -3,6 +3,8 @@
 
 import pydantic
 
+from datahub.configuration.common import ConfigurationWarning
+
 _T = TypeVar("_T")
 
 
@@ -25,8 +27,8 @@ def _validate_field_rename(cls: Type, values: dict) -> dict:
             else:
                 if print_warning:
                     warnings.warn(
-                        f"The {old_name} is deprecated, please use {new_name} instead.",
-                        UserWarning,
+                        f"{old_name} is deprecated, please use {new_name} instead.",
+                        ConfigurationWarning,
                         stacklevel=2,
                     )
                 values[new_name] = transform(values.pop(old_name))

metadata-ingestion/src/datahub/ingestion/source/aws/glue.py

@@ -51,7 +51,7 @@
 from datahub.ingestion.api.workunit import MetadataWorkUnit
 from datahub.ingestion.source.aws import s3_util
 from datahub.ingestion.source.aws.aws_common import AwsSourceConfig
-from datahub.ingestion.source.aws.s3_util import make_s3_urn
+from datahub.ingestion.source.aws.s3_util import is_s3_uri, make_s3_urn
 from datahub.ingestion.source.glue_profiling_config import GlueProfilingConfig
 from datahub.ingestion.source.state.checkpoint import Checkpoint
 from datahub.ingestion.source.state.sql_common_state import (
@@ -709,7 +709,7 @@ def get_lineage_if_enabled(
             ] = mce_builder.get_aspect_if_available(mce, DatasetPropertiesClass)
             if dataset_properties and "Location" in dataset_properties.customProperties:
                 location = dataset_properties.customProperties["Location"]
-                if location.startswith("s3://"):
+                if is_s3_uri(location):
                     s3_dataset_urn = make_s3_urn(location, self.source_config.env)
                     if self.source_config.glue_s3_lineage_direction == "upstream":
                         upstream_lineage = UpstreamLineageClass(

metadata-ingestion/src/datahub/ingestion/source/kafka.py

@@ -20,7 +20,6 @@
 from datahub.configuration.kafka import KafkaConsumerConnectionConfig
 from datahub.configuration.source_common import DatasetSourceConfigBase
 from datahub.emitter.mce_builder import (
-    DEFAULT_ENV,
     make_data_platform_urn,
     make_dataplatform_instance_urn,
     make_dataset_urn_with_platform_instance,
@@ -72,8 +71,6 @@ class KafkaTopicConfigKeys(str, Enum):
 
 
 class KafkaSourceConfig(StatefulIngestionConfigBase, DatasetSourceConfigBase):
-    env: str = DEFAULT_ENV
-    # TODO: inline the connection config
     connection: KafkaConsumerConnectionConfig = KafkaConsumerConnectionConfig()
 
     topic_patterns: AllowDenyPattern = AllowDenyPattern(allow=[".*"], deny=["^_.*"])

metadata-ingestion/src/datahub/ingestion/source/openapi.py

@@ -102,11 +102,6 @@ def get_swagger(self) -> Dict:
         return sw_dict
 
 
-# class ParserWarning(UserWarning):
-#     def __init__(self, message: str, key: str) -> None:
-#         self.message
-
-
 class ApiWorkUnit(MetadataWorkUnit):
     pass
 

metadata-ingestion/src/datahub/ingestion/source/source_registry.py

@@ -1,5 +1,6 @@
 import warnings
 
+from datahub.configuration.common import ConfigurationWarning
 from datahub.ingestion.api.registry import PluginRegistry
 from datahub.ingestion.api.source import Source
 
@@ -14,14 +15,16 @@
     "snowflake-beta",
     "snowflake",
     lambda: warnings.warn(
-        UserWarning("source type snowflake-beta is deprecated, use snowflake instead")
+        "source type snowflake-beta is deprecated, use snowflake instead",
+        ConfigurationWarning,
     ),
 )
 source_registry.register_alias(
     "bigquery-beta",
     "bigquery",
     lambda: warnings.warn(
-        UserWarning("source type bigquery-beta is deprecated, use bigquery instead")
+        "source type bigquery-beta is deprecated, use bigquery instead",
+        ConfigurationWarning,
     ),
 )
 

metadata-ingestion/src/datahub/ingestion/source/sql/mssql.py

@@ -57,11 +57,6 @@ class SQLServerConfig(BasicSQLAlchemyConfig):
         description="database (catalog). If set to Null, all databases will be considered for ingestion.",
     )
 
-    database_alias: Optional[str] = Field(
-        default=None,
-        description="Alias to apply to database when ingesting. Ignored when `database` is not set.",
-    )
-
     @pydantic.validator("uri_args")
     def passwords_match(cls, v, values, **kwargs):
         if values["use_odbc"] and "driver" not in v:

metadata-ingestion/src/datahub/ingestion/source/sql/sql_common.py

@@ -30,6 +30,7 @@
 from sqlalchemy.types import TypeDecorator, TypeEngine
 
 from datahub.configuration.common import AllowDenyPattern
+from datahub.configuration.pydantic_field_deprecation import pydantic_field_deprecated
 from datahub.emitter.mce_builder import (
     make_container_urn,
     make_data_platform_urn,
@@ -308,15 +309,20 @@ class BasicSQLAlchemyConfig(SQLAlchemyConfig):
         description="URI of database to connect to. See https://docs.sqlalchemy.org/en/14/core/engines.html#database-urls. Takes precedence over other connection parameters.",
     )
 
+    _database_alias_deprecation = pydantic_field_deprecated(
+        "database_alias",
+        message="database_alias is deprecated. Use platform_instance instead.",
+    )
+
     def get_sql_alchemy_url(self, uri_opts: Optional[Dict[str, Any]] = None) -> str:
         if not ((self.host_port and self.scheme) or self.sqlalchemy_uri):
             raise ValueError("host_port and schema or connect_uri required.")
 
         return self.sqlalchemy_uri or make_sqlalchemy_uri(
-            self.scheme,  # type: ignore
+            self.scheme,
             self.username,
             self.password.get_secret_value() if self.password is not None else None,
-            self.host_port,  # type: ignore
+            self.host_port,
             self.database,
             uri_opts=uri_opts,
         )

metadata-ingestion/tests/unit/test_airflow.py

@@ -74,6 +74,7 @@ def test_airflow_provider_info():
     assert get_provider_info()
 
 
+@pytest.mark.filterwarnings("ignore:.*is deprecated.*")
 def test_dags_load_with_no_errors(pytestconfig: pytest.Config) -> None:
     airflow_examples_folder = (
         pytestconfig.rootpath / "src/datahub_provider/example_dags"

metadata-ingestion/tests/unit/test_plugin_system.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from datahub.configuration.common import ConfigurationError
+from datahub.configuration.common import ConfigurationError, ConfigurationWarning
 from datahub.ingestion.api.registry import PluginRegistry
 from datahub.ingestion.api.sink import Sink
 from datahub.ingestion.extractor.extractor_registry import extractor_registry
@@ -97,9 +97,9 @@ class DummyClass:
         "console-alias",
         "console",
         lambda: warnings.warn(
-            UserWarning("console-alias is deprecated, use console instead")
+            ConfigurationWarning("console-alias is deprecated, use console instead")
         ),
     )
-    with pytest.warns(UserWarning):
+    with pytest.warns(ConfigurationWarning):
         assert fake_registry.get("console-alias") == ConsoleSink
     assert "console-alias" not in fake_registry.summary(verbose=False)