Feat: Add support for dbt sources external location defined in meta

sqlmesh/dbt/builtin.py

@@ -211,7 +211,6 @@ def source(package: str, name: str) -> t.Optional[BaseRelation]:
  if relation_info is None:
  logger.debug("Could not resolve source package='%s' name='%s'", package, name)
  return None
-
  return _relation_info_to_relation(relation_info, api.Relation, api.quote_policy)
 
  return source

sqlmesh/dbt/source.py

@@ -44,6 +44,7 @@ class SourceConfig(GeneralConfig):
  loaded_at_field: t.Optional[str] = None
  quoting: t.Dict[str, t.Optional[bool]] = {}
  external: t.Optional[t.Dict[str, t.Any]] = {}
+ source_meta: t.Optional[t.Dict[str, t.Any]] = {}
  columns: t.Dict[str, ColumnConfig] = {}
 
  _canonical_name: t.Optional[str] = None
@@ -86,12 +87,20 @@ def canonical_name(self, context: DbtContext) -> str:
 
  @property
  def relation_info(self) -> AttributeDict:
+ extras = {}
+ external_location = (
+ self.source_meta.get("external_location", None) if self.source_meta else None
+ )
+ if external_location:
+ extras["external"] = external_location.replace("{name}", self.table_name)
+
  return AttributeDict(
  {
  "database": self.database,
  "schema": self.schema_,
  "identifier": self.table_name,
  "type": RelationType.External.value,
  "quote_policy": AttributeDict(self.quoting),
+ **extras,
  }
  )

tests/dbt/test_config.py

@@ -340,9 +340,11 @@ def test_variables(assert_exp_eq, sushi_test_project):
 def test_source_config(sushi_test_project: Project):
  source_configs = sushi_test_project.packages["sushi"].sources
  assert set(source_configs) == {
- "streaming.items",
  "streaming.orders",
+ "parquet_file.items",
  "streaming.order_items",
+ "streaming.items",
+ "parquet_file.orders",
  }
 
  expected_config = {
@@ -359,6 +361,11 @@ def test_source_config(sushi_test_project: Project):
  == "raw.order_items"
  )
 
+ assert (
+ source_configs["parquet_file.orders"].canonical_name(sushi_test_project.context)
+ == "read_parquet('path/to/external/orders.parquet')"
+ )
+
 
 def test_seed_config(sushi_test_project: Project, mocker: MockerFixture):
  seed_configs = sushi_test_project.packages["sushi"].seeds

tests/dbt/test_manifest.py

@@ -8,6 +8,7 @@
 from sqlmesh.dbt.context import DbtContext
 from sqlmesh.dbt.manifest import ManifestHelper
 from sqlmesh.dbt.profile import Profile
+from sqlmesh.dbt.builtin import Api, _relation_info_to_relation
 from sqlmesh.utils.jinja import MacroReference
 
 pytestmark = pytest.mark.dbt
@@ -151,3 +152,36 @@ def test_variable_override():
  variable_overrides={"top_waiters:limit": 1, "start": "2020-01-01"},
  )
  assert helper.models()["top_waiters"].limit_value == 1
+
+
+@pytest.mark.xdist_group("dbt_manifest")
+def test_source_meta_external_location():
+ project_path = Path("tests/fixtures/dbt/sushi_test")
+ profile = Profile.load(DbtContext(project_path))
+
+ helper = ManifestHelper(
+ project_path,
+ project_path,
+ "sushi",
+ profile.target,
+ variable_overrides={"start": "2020-01-01"},
+ )
+
+ sources = helper.sources()
+ parquet_orders = sources["parquet_file.orders"]
+ assert parquet_orders.source_meta == {
+ "external_location": "read_parquet('path/to/external/{name}.parquet')"
+ }
+ assert (
+ parquet_orders.relation_info.external == "read_parquet('path/to/external/orders.parquet')"
+ )
+
+ api = Api("duckdb")
+ relation_info = sources["parquet_file.items"].relation_info
+ assert relation_info.external == "read_parquet('path/to/external/items.parquet')"
+
+ relation = _relation_info_to_relation(
+ sources["parquet_file.items"].relation_info, api.Relation, api.quote_policy
+ )
+ assert relation.identifier == "items"
+ assert relation.render() == "read_parquet('path/to/external/items.parquet')"

tests/fixtures/dbt/sushi_test/models/schema.yml

@@ -23,3 +23,10 @@ sources:
  - name: items
  - name: orders
  - name: order_items
+
+ - name: parquet_file
+ meta:
+ external_location: "read_parquet('path/to/external/{name}.parquet')"
+ tables:
+ - name: items
+ - name: orders