Merge main into feature/decouple-adapters-from-core

.changes/unreleased/Features-20231116-234049.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support --empty flag for schema-only dry runs
+time: 2023-11-16T23:40:49.96651-05:00
+custom:
+  Author: michelleark
+  Issue: "8971"

.changes/unreleased/Fixes-20231127-165244.yaml

@@ -0,0 +1,7 @@
+kind: Fixes
+body: Ensure we produce valid jsonschema schemas for manifest, catalog, run-results,
+  and sources
+time: 2023-11-27T16:52:44.590313-08:00
+custom:
+  Author: QMalcolm
+  Issue: "8991"

core/dbt/adapters/base/relation.py

@@ -32,6 +32,7 @@
     include_policy: Policy = field(default_factory=lambda: Policy())
     quote_policy: Policy = field(default_factory=lambda: Policy())
     dbt_created: bool = False
+    limit: Optional[int] = None
 
     # register relation types that can be renamed for the purpose of replacing relations using stages and backups
     # adding a relation type here also requires defining the associated rename macro
@@ -96,7 +97,7 @@
 
         if not search:
             # nothing was passed in
             raise dbt.common.exceptions.DbtRuntimeError(
                 "Tried to match relation, but no search path was passed!"
             )
 
@@ -190,6 +191,15 @@
         # if there is nothing set, this will return the empty string.
         return ".".join(part for _, part in self._render_iterator() if part is not None)
 
+    def render_limited(self) -> str:
+        rendered = self.render()
+        if self.limit is None:
+            return rendered
+        elif self.limit == 0:
+            return f"(select * from {rendered} where false limit 0) _dbt_limit_subq"
+        else:
+            return f"(select * from {rendered} limit {self.limit}) _dbt_limit_subq"
+
     def quoted(self, identifier):
         return "{quote_char}{identifier}{quote_char}".format(
             quote_char=self.quote_character,
@@ -204,12 +214,14 @@
     def create_ephemeral_from(
         cls: Type[Self],
         relation_config: RelationConfig,
+        limit: Optional[int],
     ) -> Self:
         # Note that ephemeral models are based on the name.
         identifier = cls.add_ephemeral_prefix(relation_config.name)
         return cls.create(
             type=cls.CTE,
             identifier=identifier,
+            limit=limit,
         ).quote(identifier=False)
 
     @classmethod
@@ -223,7 +235,6 @@
 
         config_quoting = relation_config.quoting_dict
         config_quoting.pop("column", None)
-
         # precedence: kwargs quoting > relation config quoting > base quoting > default quoting
         quote_policy = deep_merge(
             cls.get_default_quote_policy().to_dict(omit_none=True),
@@ -276,7 +287,7 @@
         return hash(self.render())
 
     def __str__(self) -> str:
-        return self.render()
+        return self.render() if self.limit is None else self.render_limited()
 
     @property
     def database(self) -> Optional[str]:
@@ -349,7 +360,7 @@
 
     def __post_init__(self):
         if not isinstance(self.information_schema_view, (type(None), str)):
             raise dbt.common.exceptions.CompilationError(
                 "Got an invalid name: {}".format(self.information_schema_view)
             )
 

core/dbt/cli/main.py

@@ -342,6 +342,7 @@ def docs_serve(ctx, **kwargs):
 @p.profile
 @p.profiles_dir
 @p.project_dir
+@p.empty
 @p.select
 @p.selector
 @p.inline
@@ -599,6 +600,7 @@ def parse(ctx, **kwargs):
 @p.profile
 @p.profiles_dir
 @p.project_dir
+@p.empty
 @p.select
 @p.selector
 @p.state

core/dbt/cli/params.py

@@ -90,6 +90,12 @@
     is_flag=True,
 )
 
+empty = click.option(
+    "--empty/--no-empty",
+    envvar="DBT_EMPTY",
+    help="If specified, limit input refs and sources to zero rows.",
+    is_flag=True,
+)
 
 enable_legacy_logger = click.option(
     "--enable-legacy-logger/--no-enable-legacy-logger",

core/dbt/context/providers.py

@@ -43,7 +43,12 @@
 from dbt.contracts.graph.metrics import MetricReference, ResolvedMetricReference
 from dbt.contracts.graph.unparsed import NodeVersion
 from dbt.common.events.functions import get_metadata_vars
-from dbt.common.exceptions import DbtInternalError, DbtRuntimeError, DbtValidationError
+from dbt.common.exceptions import (
+    DbtInternalError,
+    DbtRuntimeError,
+    DbtValidationError,
+    MacrosSourcesUnWriteableError,
+)
 from dbt.adapters.exceptions import MissingConfigError
 from dbt.exceptions import (
     CompilationError,
@@ -66,7 +71,6 @@
     TargetNotFoundError,
     DbtReferenceError,
 )
-from dbt.common.exceptions.macros import MacrosSourcesUnWriteableError
 from dbt.config import IsFQNResource
 from dbt.node_types import NodeType, ModelLanguage
 
@@ -209,6 +213,10 @@ def current_project(self):
     def Relation(self):
         return self.db_wrapper.Relation
 
+    @property
+    def resolve_limit(self) -> Optional[int]:
+        return 0 if getattr(self.config.args, "EMPTY", False) else None
+
     @abc.abstractmethod
     def __call__(self, *args: str) -> Union[str, RelationProxy, MetricReference]:
         pass
@@ -524,9 +532,9 @@ def resolve(
     def create_relation(self, target_model: ManifestNode) -> RelationProxy:
         if target_model.is_ephemeral_model:
             self.model.set_cte(target_model.unique_id, None)
-            return self.Relation.create_ephemeral_from(target_model)
+            return self.Relation.create_ephemeral_from(target_model, limit=self.resolve_limit)
         else:
-            return self.Relation.create_from(self.config, target_model)
+            return self.Relation.create_from(self.config, target_model, limit=self.resolve_limit)
 
     def validate(
         self,
@@ -583,7 +591,7 @@ def resolve(self, source_name: str, table_name: str):
                 target_kind="source",
                 disabled=(isinstance(target_source, Disabled)),
             )
-        return self.Relation.create_from(self.config, target_source)
+        return self.Relation.create_from(self.config, target_source, limit=self.resolve_limit)
 
 
 # metric` implementations

core/dbt/contracts/graph/manifest.py

@@ -1551,7 +1551,7 @@ def __init__(self, macros) -> None:
 
 
 @dataclass
-@schema_version("manifest", 11)
+@schema_version("manifest", 12)
 class WritableManifest(ArtifactMixin):
     nodes: Mapping[UniqueID, ManifestNode] = field(
         metadata=dict(description=("The nodes defined in the dbt project and its dependencies"))
@@ -1619,6 +1619,7 @@ def compatible_previous_versions(cls) -> Iterable[Tuple[str, int]]:
             ("manifest", 8),
             ("manifest", 9),
             ("manifest", 10),
+            ("manifest", 11),
         ]
 
     @classmethod

core/dbt/contracts/results.py

@@ -250,7 +250,7 @@ def write(self, path: str):
 
 
 @dataclass
-@schema_version("run-results", 5)
+@schema_version("run-results", 6)
 class RunResultsArtifact(ExecutionResult, ArtifactMixin):
     results: Sequence[RunResultOutput]
     args: Dict[str, Any] = field(default_factory=dict)
@@ -276,6 +276,7 @@ def from_execution_results(
     def compatible_previous_versions(cls) -> Iterable[Tuple[str, int]]:
         return [
             ("run-results", 4),
+            ("run-results", 5),
         ]
 
     @classmethod

core/dbt/contracts/util.py

@@ -20,6 +20,7 @@
     ValidationError,
 )
 from mashumaro.jsonschema import build_json_schema
+from mashumaro.jsonschema.dialects import DRAFT_2020_12
 import functools
 
 
@@ -194,7 +195,7 @@
     @classmethod
     @functools.lru_cache
     def json_schema(cls) -> Dict[str, Any]:
-        json_schema_obj = build_json_schema(cls, all_refs=True)
+        json_schema_obj = build_json_schema(cls, dialect=DRAFT_2020_12, with_dialect_uri=True)
         json_schema = json_schema_obj.to_dict()
         json_schema["$id"] = str(cls.dbt_schema_version)
         return json_schema
@@ -261,7 +262,7 @@
 def get_artifact_schema_version(dct: dict) -> int:
     schema_version = dct.get("metadata", {}).get("dbt_schema_version", None)
     if not schema_version:
         raise ValueError("Artifact is missing schema version")
 
     # schema_version is in this format: https://schemas.getdbt.com/dbt/manifest/v10.json
     # What the code below is doing:

core/dbt/deps/git.py

@@ -58,7 +58,7 @@
     def to_dict(self) -> Dict[str, str]:
         git_scrubbed = scrub_secrets(self.git_unrendered, env_secrets())
         if self.git_unrendered != git_scrubbed:
             warn_or_error(DepsScrubbedPackageName(package_name=git_scrubbed))
         ret = {
             "git": git_scrubbed,
             "revision": self.revision,
@@ -102,11 +102,14 @@
         self, project: Project, renderer: PackageRenderer
     ) -> ProjectPackageMetadata:
         path = self._checkout()
-        # overwrite 'revision' with actual commit SHA
-        self.revision = git.get_current_sha(path)
 
+        # raise warning (or error) if this package is not pinned
         if (self.revision == "HEAD" or self.revision in ("main", "master")) and self.warn_unpinned:
-            warn_or_error(DepsUnpinned(git=self.git))
+            warn_or_error(DepsUnpinned(revision=self.revision, git=self.git))
+
+        # now overwrite 'revision' with actual commit SHA
+        self.revision = git.get_current_sha(path)
+
         partial = PartialProject.from_project_root(path)
         return partial.render_package_metadata(renderer)