a06f331eb8
CI / benchmark (push) Has been skipped
install-script / posix-syntax (push) Successful in 6m1s
CI / build-onnx (push) Failing after 6m43s
init-smoke / dry-run (push) Failing after 15m57s
security / govulncheck (push) Has been cancelled
security / trivy-fs (push) Has been cancelled
CI / test (1.26, ubuntu-latest) (push) Has been cancelled
Scorecard supply-chain security / Scorecard analysis (push) Has been cancelled
CI / test (1.26, macos-latest) (push) Has been cancelled
CI / build-windows (push) Has been cancelled
CI / lint (push) Has been cancelled
install-script / powershell-syntax (push) Has been cancelled
install-script / install (macos-14) (push) Has been cancelled
install-script / install (ubuntu-latest) (push) Has been cancelled
893 lines
29 KiB
Go
893 lines
29 KiB
Go
package languages
|
|
|
|
import (
|
|
"maps"
|
|
"path/filepath"
|
|
"regexp"
|
|
"strings"
|
|
|
|
"gopkg.in/yaml.v3"
|
|
|
|
"github.com/zzet/gortex/internal/graph"
|
|
"github.com/zzet/gortex/internal/parser"
|
|
"github.com/zzet/gortex/internal/sql"
|
|
)
|
|
|
|
// dbt / SQLMesh column-level search (roadmap B10).
|
|
//
|
|
// Three ingest paths land here, each dispatched into by a host
|
|
// extractor that already owns the file extension:
|
|
//
|
|
// - dbt SQL models — Jinja-templated `.sql` under a dbt
|
|
// project's models/ tree. Dispatched from
|
|
// SQLExtractor.Extract via classifySQLFile.
|
|
// - SQLMesh SQL models — `.sql` files that open with a `MODEL (…)`
|
|
// block. Same dispatch point.
|
|
// - dbt schema YAML — `schema.yml` / properties files declaring
|
|
// `models:` / `sources:` / `seeds:` /
|
|
// `snapshots:` with their columns. Dispatched
|
|
// from YAMLExtractor.Extract.
|
|
//
|
|
// Graph shape. Every model / seed / snapshot / source becomes a
|
|
// KindTable node; every declared or projected column becomes a
|
|
// KindColumn node linked to its model with EdgeMemberOf — that is the
|
|
// "column-level search" surface (search_symbols kind:column). Model
|
|
// lineage — dbt `ref()` / `source()`, SQLMesh `FROM` / `JOIN` — becomes
|
|
// EdgeDependsOn edges between the KindTable nodes. IDs are deterministic
|
|
// and name-derived so lineage edges resolve through node identity
|
|
// without a resolver pass:
|
|
//
|
|
// dbt::model::<name> model, seed, or snapshot (the dbt
|
|
// `ref()` namespace is flat across
|
|
// all three resource types)
|
|
// dbt::source::<source>.<table> dbt source (the `source()` namespace)
|
|
// sqlmesh::model::<qualified> SQLMesh model, keyed by its declared
|
|
// (schema-qualified) name
|
|
// <model-id>::<column> a column of the model above
|
|
var (
|
|
// sqlmeshModelRe matches a SQLMesh model file: a `MODEL (` block
|
|
// at the start of the file, tolerating leading line / block
|
|
// comments. The match end sits just past the opening paren.
|
|
sqlmeshModelRe = regexp.MustCompile(`(?is)^\s*(?:--[^\n]*\n\s*|/\*.*?\*/\s*)*MODEL\s*\(`)
|
|
// dbtJinjaMarkerRe recognises the load-bearing dbt Jinja calls.
|
|
// Any of these in a `.sql` file is a definitive dbt-model signal.
|
|
dbtJinjaMarkerRe = regexp.MustCompile(`(?is)\{\{[-\s]*(?:ref|source|config|this)\b|\{%[-\s]*(?:snapshot|materialization)\b`)
|
|
|
|
dbtRefRe = regexp.MustCompile(`(?is)\{\{[-\s]*ref\s*\(\s*('[^']+'|"[^"]+")\s*(?:,\s*('[^']+'|"[^"]+")\s*)?\)[-\s]*\}\}`)
|
|
dbtSourceRe = regexp.MustCompile(`(?is)\{\{[-\s]*source\s*\(\s*('[^']+'|"[^"]+")\s*,\s*('[^']+'|"[^"]+")\s*\)[-\s]*\}\}`)
|
|
dbtConfigRe = regexp.MustCompile(`(?is)\{\{[-\s]*config\s*\((.*?)\)[-\s]*\}\}`)
|
|
dbtSnapshotRe = regexp.MustCompile(`(?i)\{%[-\s]*snapshot\s+([A-Za-z_]\w*)\s*[-\s]*%\}`)
|
|
|
|
jinjaExprRe = regexp.MustCompile(`(?s)\{\{.*?\}\}`)
|
|
jinjaStmtRe = regexp.MustCompile(`(?s)\{%.*?%\}`)
|
|
jinjaCommentRe = regexp.MustCompile(`(?s)\{#.*?#\}`)
|
|
|
|
sqlLineCommentRe = regexp.MustCompile(`(?m)--[^\n]*`)
|
|
sqlBlockCommentRe = regexp.MustCompile(`(?s)/\*.*?\*/`)
|
|
sqlmeshMacroRe = regexp.MustCompile(`@\w+`)
|
|
|
|
// dbtConfigKVRe pulls a single `key='value'` / `key="value"` pair
|
|
// out of a config(...) argument blob.
|
|
dbtConfigKVRe = regexp.MustCompile(`(?i)\b(materialized|schema|alias|database|incremental_strategy|unique_key)\s*=\s*('[^']*'|"[^"]*")`)
|
|
)
|
|
|
|
// classifySQLFile decides whether a `.sql` file is a dbt model, a
|
|
// SQLMesh model, or neither. Returns "dbt", "sqlmesh", or "".
|
|
//
|
|
// Order matters: the SQLMesh `MODEL (` block is structurally
|
|
// unambiguous so it is checked first; dbt Jinja markers are next; a
|
|
// path-based fallback then catches pure-SQL dbt models (a dbt model
|
|
// need not contain any Jinja) — but only when the file actually opens
|
|
// with a query, so a stray DDL script under a models/ directory is not
|
|
// misclassified.
|
|
func classifySQLFile(filePath string, src []byte) string {
|
|
if sqlmeshModelRe.Match(src) {
|
|
return "sqlmesh"
|
|
}
|
|
if dbtJinjaMarkerRe.Match(src) {
|
|
return "dbt"
|
|
}
|
|
if isDbtModelPath(filePath) && opensWithQuery(src) {
|
|
return "dbt"
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// isDbtModelPath reports whether filePath sits in the conventional dbt
|
|
// models/ or snapshots/ tree. Seeds are `.csv` and never reach the SQL
|
|
// extractor; sources have no SQL file at all.
|
|
func isDbtModelPath(filePath string) bool {
|
|
lower := strings.ToLower(filepath.ToSlash(filePath))
|
|
if strings.Contains(lower, "/models/") || strings.Contains(lower, "/snapshots/") {
|
|
return true
|
|
}
|
|
return strings.HasPrefix(lower, "models/") || strings.HasPrefix(lower, "snapshots/")
|
|
}
|
|
|
|
// opensWithQuery reports whether the first significant token of a SQL
|
|
// file is `SELECT`, `WITH`, or a Jinja delimiter — the shapes a dbt
|
|
// model file takes. A file that opens with `CREATE` / `ALTER` / `INSERT`
|
|
// / etc. is a DDL or DML script, not a dbt model.
|
|
func opensWithQuery(src []byte) bool {
|
|
s := strings.TrimSpace(stripSQLComments(string(src)))
|
|
if s == "" {
|
|
return false
|
|
}
|
|
up := strings.ToUpper(s)
|
|
return strings.HasPrefix(up, "SELECT") || strings.HasPrefix(up, "WITH") ||
|
|
strings.HasPrefix(s, "{{") || strings.HasPrefix(s, "{%") || strings.HasPrefix(s, "(")
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// dbt SQL models
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// extractDbtSQLModel emits the graph fragment for a single dbt SQL
|
|
// model file. The host SQLExtractor has already appended the file node
|
|
// (fileID); this function adds the model KindTable node, its columns,
|
|
// and its `ref()` / `source()` lineage.
|
|
func extractDbtSQLModel(filePath, fileID string, src []byte, result *parser.ExtractionResult) {
|
|
resourceType := "model"
|
|
name := dbtModelNameFromPath(filePath)
|
|
if m := dbtSnapshotRe.FindSubmatch(src); m != nil {
|
|
resourceType = "snapshot"
|
|
name = string(m[1])
|
|
}
|
|
if name == "" {
|
|
return
|
|
}
|
|
|
|
modelID := dbtModelID(name)
|
|
endLine := lineCount(src)
|
|
meta := map[string]any{
|
|
"framework": "dbt",
|
|
"sql_type": "dbt_model",
|
|
"resource_type": resourceType,
|
|
"model": name,
|
|
}
|
|
cfg := parseDbtConfig(src)
|
|
materialized := cfg["materialized"]
|
|
if materialized == "" {
|
|
if resourceType == "snapshot" {
|
|
materialized = "snapshot"
|
|
} else {
|
|
materialized = "view"
|
|
}
|
|
}
|
|
meta["materialized"] = materialized
|
|
if v := cfg["schema"]; v != "" {
|
|
meta["schema"] = v
|
|
}
|
|
if v := cfg["alias"]; v != "" {
|
|
meta["alias"] = v
|
|
}
|
|
if v := cfg["database"]; v != "" {
|
|
meta["database"] = v
|
|
}
|
|
|
|
result.Nodes = append(result.Nodes, &graph.Node{
|
|
ID: modelID, Kind: graph.KindTable, Name: name,
|
|
QualName: "dbt.model." + name,
|
|
FilePath: filePath, StartLine: 1, EndLine: endLine,
|
|
Language: "sql", Meta: meta,
|
|
})
|
|
result.Edges = append(result.Edges, &graph.Edge{
|
|
From: fileID, To: modelID, Kind: graph.EdgeDefines,
|
|
FilePath: filePath, Line: 1,
|
|
})
|
|
|
|
emitDbtLineage(filePath, modelID, src, result)
|
|
emitDbtModelColumns(filePath, modelID, name, deJinja(src), result)
|
|
}
|
|
|
|
// emitDbtLineage walks every `{{ ref(...) }}` and `{{ source(...) }}`
|
|
// call in src and emits an EdgeDependsOn from the model to the upstream
|
|
// node. `source()` references also materialise a minimal KindTable
|
|
// source node so the edge always lands on a real node even when no
|
|
// schema YAML declares the source; a later schema-YAML pass enriches it.
|
|
func emitDbtLineage(filePath, modelID string, src []byte, result *parser.ExtractionResult) {
|
|
seen := make(map[string]bool)
|
|
emitDep := func(targetID, link, detail string, line int) {
|
|
key := targetID + "\x00" + link
|
|
if seen[key] {
|
|
return
|
|
}
|
|
seen[key] = true
|
|
m := map[string]any{"link": link}
|
|
if detail != "" {
|
|
m["ref_package"] = detail
|
|
}
|
|
result.Edges = append(result.Edges, &graph.Edge{
|
|
From: modelID, To: targetID, Kind: graph.EdgeDependsOn,
|
|
FilePath: filePath, Line: line, Meta: m,
|
|
})
|
|
}
|
|
|
|
for _, m := range dbtRefRe.FindAllSubmatchIndex(src, -1) {
|
|
arg1 := unquoteSQL(string(src[m[2]:m[3]]))
|
|
pkg := ""
|
|
refName := arg1
|
|
if m[4] >= 0 { // two-arg form: ref('package', 'model')
|
|
pkg = arg1
|
|
refName = unquoteSQL(string(src[m[4]:m[5]]))
|
|
}
|
|
if refName == "" {
|
|
continue
|
|
}
|
|
emitDep(dbtModelID(refName), "ref", pkg, lineAt(src, m[0]))
|
|
}
|
|
|
|
for _, m := range dbtSourceRe.FindAllSubmatchIndex(src, -1) {
|
|
sourceName := unquoteSQL(string(src[m[2]:m[3]]))
|
|
tableName := unquoteSQL(string(src[m[4]:m[5]]))
|
|
if sourceName == "" || tableName == "" {
|
|
continue
|
|
}
|
|
srcID := dbtSourceID(sourceName, tableName)
|
|
line := lineAt(src, m[0])
|
|
ensureDbtSourceNode(filePath, srcID, sourceName, tableName, line, result)
|
|
emitDep(srcID, "source", "", line)
|
|
}
|
|
}
|
|
|
|
// emitDbtModelColumns records a model's own output columns. dejinjaed
|
|
// is the model body with Jinja delimiters neutralised so the SQL
|
|
// projection parser can read the final SELECT.
|
|
func emitDbtModelColumns(filePath, modelID, modelName, dejinjaed string, result *parser.ExtractionResult) {
|
|
cols := sql.ProjectionColumns(stripSQLComments(dejinjaed))
|
|
for _, col := range cols {
|
|
emitDbtColumn(filePath, modelID, modelName, col, "sql", nil, result)
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// SQLMesh SQL models
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// extractSQLMeshSQLModel emits the graph fragment for a SQLMesh `.sql`
|
|
// model: the `MODEL (…)` block's properties become node Meta, the
|
|
// `columns (…)` property (or the body's final SELECT projection when
|
|
// absent) becomes KindColumn nodes, and the body's FROM / JOIN table
|
|
// references become EdgeDependsOn lineage edges.
|
|
func extractSQLMeshSQLModel(filePath, fileID string, src []byte, result *parser.ExtractionResult) {
|
|
props, columnDefs, bodyStart, ok := parseSQLMeshModel(src)
|
|
if !ok {
|
|
return
|
|
}
|
|
name := props["name"]
|
|
if name == "" {
|
|
name = dbtModelNameFromPath(filePath)
|
|
}
|
|
if name == "" {
|
|
return
|
|
}
|
|
|
|
modelID := sqlmeshModelID(name)
|
|
meta := map[string]any{
|
|
"framework": "sqlmesh",
|
|
"sql_type": "sqlmesh_model",
|
|
"model": name,
|
|
}
|
|
if i := strings.LastIndex(name, "."); i >= 0 {
|
|
meta["schema"] = name[:i]
|
|
}
|
|
if v := props["kind"]; v != "" {
|
|
// `kind` can be `FULL` or `INCREMENTAL_BY_TIME_RANGE(...)`; the
|
|
// leading token is the stable classifier.
|
|
meta["materialized"] = strings.ToLower(firstToken(v))
|
|
meta["kind"] = v
|
|
} else {
|
|
meta["materialized"] = "unknown"
|
|
}
|
|
for _, k := range []string{"cron", "grain", "owner", "dialect", "description"} {
|
|
if v := props[k]; v != "" {
|
|
meta[k] = unquoteSQL(v)
|
|
}
|
|
}
|
|
|
|
result.Nodes = append(result.Nodes, &graph.Node{
|
|
ID: modelID, Kind: graph.KindTable, Name: sqlmeshLeafName(name),
|
|
QualName: "sqlmesh.model." + name,
|
|
FilePath: filePath, StartLine: 1, EndLine: lineCount(src),
|
|
Language: "sql", Meta: meta,
|
|
})
|
|
result.Edges = append(result.Edges, &graph.Edge{
|
|
From: fileID, To: modelID, Kind: graph.EdgeDefines,
|
|
FilePath: filePath, Line: 1,
|
|
})
|
|
|
|
// Columns: the explicit `columns (...)` property wins; otherwise
|
|
// fall back to the body's final SELECT projection.
|
|
if len(columnDefs) > 0 {
|
|
for _, cd := range columnDefs {
|
|
var cm map[string]any
|
|
if cd.dataType != "" {
|
|
cm = map[string]any{"data_type": cd.dataType}
|
|
}
|
|
emitDbtColumn(filePath, modelID, sqlmeshLeafName(name), cd.name, "model_block", cm, result)
|
|
}
|
|
} else if bodyStart < len(src) {
|
|
body := stripSQLComments(sqlmeshMacroRe.ReplaceAllString(string(src[bodyStart:]), " "))
|
|
for _, col := range sql.ProjectionColumns(body) {
|
|
emitDbtColumn(filePath, modelID, sqlmeshLeafName(name), col, "sql", nil, result)
|
|
}
|
|
}
|
|
|
|
// Lineage: every table the body reads from is an upstream model.
|
|
if bodyStart < len(src) {
|
|
body := stripSQLComments(sqlmeshMacroRe.ReplaceAllString(string(src[bodyStart:]), " "))
|
|
seen := make(map[string]bool)
|
|
for _, ref := range sql.ExtractTables(body) {
|
|
upstream := ref.Table
|
|
if ref.Schema != "" {
|
|
upstream = ref.Schema + "." + ref.Table
|
|
}
|
|
upstreamID := sqlmeshModelID(upstream)
|
|
if upstreamID == modelID || seen[upstreamID] {
|
|
continue
|
|
}
|
|
seen[upstreamID] = true
|
|
result.Edges = append(result.Edges, &graph.Edge{
|
|
From: modelID, To: upstreamID, Kind: graph.EdgeDependsOn,
|
|
FilePath: filePath, Line: 1,
|
|
Meta: map[string]any{"link": "from", "op": ref.Op},
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
// sqlmeshColumn is one entry of a SQLMesh `columns (...)` property.
|
|
type sqlmeshColumn struct {
|
|
name string
|
|
dataType string
|
|
}
|
|
|
|
// parseSQLMeshModel locates and parses the leading `MODEL (…)` block.
|
|
// Returns the top-level scalar properties, the parsed `columns (...)`
|
|
// list (nil when the property is absent), the byte offset where the
|
|
// post-block SQL body starts, and ok=false when no MODEL block is found.
|
|
func parseSQLMeshModel(src []byte) (props map[string]string, columns []sqlmeshColumn, bodyStart int, ok bool) {
|
|
loc := sqlmeshModelRe.FindIndex(src)
|
|
if loc == nil {
|
|
return nil, nil, 0, false
|
|
}
|
|
openParen := loc[1] - 1 // sqlmeshModelRe ends just past `(`
|
|
closeParen := scanMatchingParen(src, openParen)
|
|
if closeParen < 0 {
|
|
return nil, nil, 0, false
|
|
}
|
|
block := string(src[openParen+1 : closeParen])
|
|
|
|
props = make(map[string]string)
|
|
for _, prop := range splitTopLevelCommas(block) {
|
|
prop = strings.TrimSpace(prop)
|
|
if prop == "" {
|
|
continue
|
|
}
|
|
key, value := splitFirstToken(prop)
|
|
key = strings.ToLower(key)
|
|
if key == "columns" {
|
|
columns = parseSQLMeshColumns(value)
|
|
continue
|
|
}
|
|
props[key] = strings.TrimSpace(value)
|
|
}
|
|
|
|
bodyStart = closeParen + 1
|
|
for bodyStart < len(src) {
|
|
c := src[bodyStart]
|
|
if c == ';' || c == ' ' || c == '\n' || c == '\r' || c == '\t' {
|
|
bodyStart++
|
|
continue
|
|
}
|
|
break
|
|
}
|
|
return props, columns, bodyStart, true
|
|
}
|
|
|
|
// parseSQLMeshColumns parses a `columns (...)` property value — the
|
|
// outer parens still attached — into (name, type) pairs.
|
|
func parseSQLMeshColumns(value string) []sqlmeshColumn {
|
|
value = strings.TrimSpace(value)
|
|
value = strings.TrimPrefix(value, "(")
|
|
value = strings.TrimSuffix(value, ")")
|
|
var out []sqlmeshColumn
|
|
for _, entry := range splitTopLevelCommas(value) {
|
|
name, rest := splitFirstToken(strings.TrimSpace(entry))
|
|
name = strings.TrimSpace(stripSQLIdentQuotes(name))
|
|
if name == "" {
|
|
continue
|
|
}
|
|
out = append(out, sqlmeshColumn{name: name, dataType: strings.TrimSpace(rest)})
|
|
}
|
|
return out
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// dbt schema YAML
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// extractDbtSchemaYAML detects and extracts a dbt schema / properties
|
|
// YAML file — the `schema.yml`-style files that describe `models:`,
|
|
// `sources:`, `seeds:`, and `snapshots:` along with their columns.
|
|
// Returns true when the file was a dbt schema file (so the YAML
|
|
// extractor skips its generic top-level-keys fallback), false otherwise.
|
|
func extractDbtSchemaYAML(filePath, fileID string, src []byte, result *parser.ExtractionResult) bool {
|
|
var doc yaml.Node
|
|
if err := yaml.Unmarshal(src, &doc); err != nil {
|
|
return false
|
|
}
|
|
root := documentMapping(&doc)
|
|
if root == nil || !isDbtSchemaYAML(root, filePath) {
|
|
return false
|
|
}
|
|
|
|
// models / seeds / snapshots: all live in the flat dbt `ref()`
|
|
// namespace, so they share the dbt::model:: ID prefix and are
|
|
// distinguished only by Meta["resource_type"].
|
|
for _, group := range []struct{ key, resourceType, sqlType string }{
|
|
{"models", "model", "dbt_model"},
|
|
{"seeds", "seed", "dbt_seed"},
|
|
{"snapshots", "snapshot", "dbt_snapshot"},
|
|
} {
|
|
for _, item := range sequenceItems(mappingGet(root, group.key)) {
|
|
if item.Kind != yaml.MappingNode {
|
|
continue
|
|
}
|
|
name := scalarOf(mappingGet(item, "name"))
|
|
if name == "" {
|
|
continue
|
|
}
|
|
modelID := dbtModelID(name)
|
|
line := item.Line
|
|
if line <= 0 {
|
|
line = 1
|
|
}
|
|
meta := map[string]any{
|
|
"framework": "dbt",
|
|
"sql_type": group.sqlType,
|
|
"resource_type": group.resourceType,
|
|
"model": name,
|
|
}
|
|
if desc := scalarOf(mappingGet(item, "description")); desc != "" {
|
|
meta["description"] = truncateText(desc, 280)
|
|
}
|
|
if mat := dbtConfigMaterialized(item); mat != "" {
|
|
meta["materialized"] = mat
|
|
}
|
|
result.Nodes = append(result.Nodes, &graph.Node{
|
|
ID: modelID, Kind: graph.KindTable, Name: name,
|
|
QualName: "dbt.model." + name,
|
|
FilePath: filePath, StartLine: line, EndLine: line,
|
|
Language: "yaml", Meta: meta,
|
|
})
|
|
result.Edges = append(result.Edges, &graph.Edge{
|
|
From: fileID, To: modelID, Kind: graph.EdgeDefines,
|
|
FilePath: filePath, Line: line,
|
|
})
|
|
emitDbtSchemaColumns(filePath, modelID, name, item, result)
|
|
}
|
|
}
|
|
|
|
// sources: a `sources:` entry groups one or more `tables:`, each of
|
|
// which is an independently ref-able (via `source()`) relation.
|
|
for _, srcItem := range sequenceItems(mappingGet(root, "sources")) {
|
|
if srcItem.Kind != yaml.MappingNode {
|
|
continue
|
|
}
|
|
sourceName := scalarOf(mappingGet(srcItem, "name"))
|
|
if sourceName == "" {
|
|
continue
|
|
}
|
|
sourceSchema := scalarOf(mappingGet(srcItem, "schema"))
|
|
sourceDB := scalarOf(mappingGet(srcItem, "database"))
|
|
for _, tbl := range sequenceItems(mappingGet(srcItem, "tables")) {
|
|
if tbl.Kind != yaml.MappingNode {
|
|
continue
|
|
}
|
|
tableName := scalarOf(mappingGet(tbl, "name"))
|
|
if tableName == "" {
|
|
continue
|
|
}
|
|
srcID := dbtSourceID(sourceName, tableName)
|
|
line := tbl.Line
|
|
if line <= 0 {
|
|
line = 1
|
|
}
|
|
meta := map[string]any{
|
|
"framework": "dbt",
|
|
"sql_type": "dbt_source",
|
|
"resource_type": "source",
|
|
"source": sourceName,
|
|
"table": tableName,
|
|
}
|
|
if sourceSchema != "" {
|
|
meta["schema"] = sourceSchema
|
|
}
|
|
if sourceDB != "" {
|
|
meta["database"] = sourceDB
|
|
}
|
|
if desc := scalarOf(mappingGet(tbl, "description")); desc != "" {
|
|
meta["description"] = truncateText(desc, 280)
|
|
}
|
|
result.Nodes = append(result.Nodes, &graph.Node{
|
|
ID: srcID, Kind: graph.KindTable, Name: sourceName + "." + tableName,
|
|
QualName: "dbt.source." + sourceName + "." + tableName,
|
|
FilePath: filePath, StartLine: line, EndLine: line,
|
|
Language: "yaml", Meta: meta,
|
|
})
|
|
result.Edges = append(result.Edges, &graph.Edge{
|
|
From: fileID, To: srcID, Kind: graph.EdgeDefines,
|
|
FilePath: filePath, Line: line,
|
|
})
|
|
emitDbtSchemaColumns(filePath, srcID, sourceName+"."+tableName, tbl, result)
|
|
}
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
// isDbtSchemaYAML fingerprints a dbt schema / properties YAML file. It
|
|
// requires a `models` / `seeds` / `snapshots` sequence of name-bearing
|
|
// mappings (with at least one descriptive sub-key, or a top-level
|
|
// `version`), or a `sources` sequence whose entries carry `tables` —
|
|
// a shape that does not collide with dbt_project.yml (whose `models:`
|
|
// is a config *mapping*) or with arbitrary config YAML.
|
|
func isDbtSchemaYAML(root *yaml.Node, filePath string) bool {
|
|
if root == nil {
|
|
return false
|
|
}
|
|
switch strings.ToLower(filepath.Base(filePath)) {
|
|
case "dbt_project.yml", "dbt_project.yaml", "packages.yml", "packages.yaml",
|
|
"dependencies.yml", "dependencies.yaml", "profiles.yml", "profiles.yaml",
|
|
"selectors.yml", "selectors.yaml":
|
|
return false
|
|
}
|
|
hasVersion := scalarOf(mappingGet(root, "version")) != ""
|
|
for _, key := range []string{"models", "seeds", "snapshots"} {
|
|
for _, item := range sequenceItems(mappingGet(root, key)) {
|
|
if item.Kind != yaml.MappingNode {
|
|
continue
|
|
}
|
|
if scalarOf(mappingGet(item, "name")) == "" {
|
|
continue
|
|
}
|
|
if hasVersion ||
|
|
mappingGet(item, "columns") != nil ||
|
|
mappingGet(item, "description") != nil ||
|
|
mappingGet(item, "config") != nil ||
|
|
mappingGet(item, "tests") != nil ||
|
|
mappingGet(item, "data_tests") != nil {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
for _, item := range sequenceItems(mappingGet(root, "sources")) {
|
|
if item.Kind != yaml.MappingNode {
|
|
continue
|
|
}
|
|
if scalarOf(mappingGet(item, "name")) != "" && mappingGet(item, "tables") != nil {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// emitDbtSchemaColumns walks a model / source `columns:` sequence and
|
|
// emits a KindColumn node per entry, carrying description, data type,
|
|
// and the declared tests in Meta.
|
|
func emitDbtSchemaColumns(filePath, ownerID, ownerName string, item *yaml.Node, result *parser.ExtractionResult) {
|
|
for _, col := range sequenceItems(mappingGet(item, "columns")) {
|
|
if col.Kind != yaml.MappingNode {
|
|
continue
|
|
}
|
|
colName := scalarOf(mappingGet(col, "name"))
|
|
if colName == "" {
|
|
continue
|
|
}
|
|
cm := map[string]any{}
|
|
if desc := scalarOf(mappingGet(col, "description")); desc != "" {
|
|
cm["description"] = truncateText(desc, 280)
|
|
}
|
|
dataType := scalarOf(mappingGet(col, "data_type"))
|
|
if dataType == "" {
|
|
dataType = scalarOf(mappingGet(col, "type"))
|
|
}
|
|
if dataType != "" {
|
|
cm["data_type"] = dataType
|
|
}
|
|
if tests := dbtColumnTests(col); len(tests) > 0 {
|
|
cm["tests"] = tests
|
|
}
|
|
emitDbtColumn(filePath, ownerID, ownerName, colName, "schema_yaml", cm, result)
|
|
}
|
|
}
|
|
|
|
// dbtColumnTests collects the test names declared on a column. dbt
|
|
// accepts `tests:` (legacy) and `data_tests:` (1.8+); each entry is
|
|
// either a scalar test name or a single-key mapping (`relationships:`,
|
|
// `accepted_values:`, …).
|
|
func dbtColumnTests(col *yaml.Node) []string {
|
|
var out []string
|
|
for _, key := range []string{"tests", "data_tests"} {
|
|
for _, t := range sequenceItems(mappingGet(col, key)) {
|
|
switch t.Kind {
|
|
case yaml.ScalarNode:
|
|
if t.Value != "" {
|
|
out = append(out, t.Value)
|
|
}
|
|
case yaml.MappingNode:
|
|
if len(t.Content) >= 1 && t.Content[0] != nil && t.Content[0].Value != "" {
|
|
out = append(out, t.Content[0].Value)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// dbtConfigMaterialized pulls the materialization out of a schema-YAML
|
|
// `config:` block (`config: { materialized: table }`).
|
|
func dbtConfigMaterialized(item *yaml.Node) string {
|
|
cfg := mappingGet(item, "config")
|
|
if cfg == nil {
|
|
return ""
|
|
}
|
|
return scalarOf(mappingGet(cfg, "materialized"))
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// shared node emission + helpers
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// emitDbtColumn appends one KindColumn node and its EdgeMemberOf edge,
|
|
// merging the framework-agnostic Meta with any source-specific extra.
|
|
func emitDbtColumn(filePath, ownerID, ownerName, colName, source string, extra map[string]any, result *parser.ExtractionResult) {
|
|
colName = strings.TrimSpace(stripSQLIdentQuotes(colName))
|
|
if colName == "" {
|
|
return
|
|
}
|
|
colID := ownerID + "::" + colName
|
|
meta := map[string]any{
|
|
"table": ownerName,
|
|
"column": colName,
|
|
"source": source,
|
|
}
|
|
maps.Copy(meta, extra)
|
|
result.Nodes = append(result.Nodes, &graph.Node{
|
|
ID: colID, Kind: graph.KindColumn, Name: colName,
|
|
FilePath: filePath, StartLine: 1, EndLine: 1,
|
|
Language: "sql", Meta: meta,
|
|
})
|
|
result.Edges = append(result.Edges, &graph.Edge{
|
|
From: colID, To: ownerID, Kind: graph.EdgeMemberOf,
|
|
FilePath: filePath, Line: 1,
|
|
})
|
|
}
|
|
|
|
// ensureDbtSourceNode appends a minimal KindTable node for a dbt source
|
|
// referenced by `source()` but not (yet) seen in a schema YAML. The
|
|
// schema-YAML pass overwrites it with the richer node when it runs.
|
|
func ensureDbtSourceNode(filePath, srcID, sourceName, tableName string, line int, result *parser.ExtractionResult) {
|
|
for _, n := range result.Nodes {
|
|
if n.ID == srcID {
|
|
return
|
|
}
|
|
}
|
|
result.Nodes = append(result.Nodes, &graph.Node{
|
|
ID: srcID, Kind: graph.KindTable, Name: sourceName + "." + tableName,
|
|
QualName: "dbt.source." + sourceName + "." + tableName,
|
|
FilePath: filePath, StartLine: line, EndLine: line,
|
|
Language: "sql",
|
|
Meta: map[string]any{
|
|
"framework": "dbt",
|
|
"sql_type": "dbt_source",
|
|
"resource_type": "source",
|
|
"source": sourceName,
|
|
"table": tableName,
|
|
},
|
|
})
|
|
}
|
|
|
|
// dbtModelID / dbtSourceID / sqlmeshModelID centralise the synthetic
|
|
// node-ID conventions documented at the top of this file.
|
|
func dbtModelID(name string) string { return "dbt::model::" + name }
|
|
func sqlmeshModelID(name string) string { return "sqlmesh::model::" + strings.ToLower(name) }
|
|
|
|
func dbtSourceID(source, table string) string {
|
|
return "dbt::source::" + source + "." + table
|
|
}
|
|
|
|
// dbtModelNameFromPath derives a dbt model name from its file path:
|
|
// dbt keys a model by its bare filename stem.
|
|
func dbtModelNameFromPath(filePath string) string {
|
|
base := filepath.Base(filePath)
|
|
return strings.TrimSuffix(base, filepath.Ext(base))
|
|
}
|
|
|
|
// sqlmeshLeafName returns the unqualified tail of a (possibly
|
|
// schema-qualified) SQLMesh model name.
|
|
func sqlmeshLeafName(name string) string {
|
|
if i := strings.LastIndex(name, "."); i >= 0 {
|
|
return name[i+1:]
|
|
}
|
|
return name
|
|
}
|
|
|
|
// parseDbtConfig extracts the recognised keys from every
|
|
// `{{ config(...) }}` call in src. Later calls win on key collision.
|
|
func parseDbtConfig(src []byte) map[string]string {
|
|
out := map[string]string{}
|
|
for _, block := range dbtConfigRe.FindAllSubmatch(src, -1) {
|
|
for _, kv := range dbtConfigKVRe.FindAllSubmatch(block[1], -1) {
|
|
out[strings.ToLower(string(kv[1]))] = unquoteSQL(string(kv[2]))
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// deJinja neutralises Jinja delimiters in a dbt model body so the
|
|
// regex SQL parsers can read the underlying query. `ref()` / `source()`
|
|
// calls collapse to a placeholder relation identifier (keeping FROM
|
|
// clauses parseable); every other `{{ }}` / `{% %}` / `{# #}` construct
|
|
// collapses to whitespace.
|
|
func deJinja(src []byte) string {
|
|
s := jinjaCommentRe.ReplaceAllString(string(src), " ")
|
|
s = jinjaStmtRe.ReplaceAllString(s, " ")
|
|
s = dbtRefRe.ReplaceAllString(s, " __dbt_rel__ ")
|
|
s = dbtSourceRe.ReplaceAllString(s, " __dbt_rel__ ")
|
|
s = jinjaExprRe.ReplaceAllString(s, " __dbt_expr__ ")
|
|
return s
|
|
}
|
|
|
|
// stripSQLComments removes `--` line comments and `/* */` block
|
|
// comments so they cannot leak FROM-like text into the regex parsers.
|
|
func stripSQLComments(s string) string {
|
|
s = sqlBlockCommentRe.ReplaceAllString(s, " ")
|
|
s = sqlLineCommentRe.ReplaceAllString(s, "")
|
|
return s
|
|
}
|
|
|
|
// unquoteSQL strips a single matched pair of single or double quotes.
|
|
func unquoteSQL(s string) string {
|
|
s = strings.TrimSpace(s)
|
|
if len(s) >= 2 {
|
|
f, l := s[0], s[len(s)-1]
|
|
if (f == '\'' && l == '\'') || (f == '"' && l == '"') {
|
|
return s[1 : len(s)-1]
|
|
}
|
|
}
|
|
return s
|
|
}
|
|
|
|
// stripSQLIdentQuotes strips ANSI / MySQL / T-SQL identifier quoting.
|
|
func stripSQLIdentQuotes(s string) string {
|
|
s = strings.TrimSpace(s)
|
|
if len(s) >= 2 {
|
|
f, l := s[0], s[len(s)-1]
|
|
switch {
|
|
case f == '"' && l == '"', f == '`' && l == '`', f == '[' && l == ']':
|
|
return s[1 : len(s)-1]
|
|
}
|
|
}
|
|
return s
|
|
}
|
|
|
|
// firstToken returns the first whitespace-delimited token of s.
|
|
func firstToken(s string) string {
|
|
s = strings.TrimSpace(s)
|
|
if i := strings.IndexFunc(s, func(r rune) bool {
|
|
return r == ' ' || r == '\t' || r == '\n' || r == '\r' || r == '('
|
|
}); i >= 0 {
|
|
return s[:i]
|
|
}
|
|
return s
|
|
}
|
|
|
|
// splitFirstToken splits s into its first whitespace-delimited token
|
|
// and the trimmed remainder.
|
|
func splitFirstToken(s string) (token, rest string) {
|
|
s = strings.TrimSpace(s)
|
|
i := strings.IndexFunc(s, func(r rune) bool {
|
|
return r == ' ' || r == '\t' || r == '\n' || r == '\r'
|
|
})
|
|
if i < 0 {
|
|
return s, ""
|
|
}
|
|
return s[:i], strings.TrimSpace(s[i+1:])
|
|
}
|
|
|
|
// splitTopLevelCommas splits s on commas that sit at paren depth zero
|
|
// and outside single/double-quoted string literals.
|
|
func splitTopLevelCommas(s string) []string {
|
|
var out []string
|
|
depth := 0
|
|
var quote byte
|
|
cur := strings.Builder{}
|
|
for i := 0; i < len(s); i++ {
|
|
c := s[i]
|
|
if quote != 0 {
|
|
cur.WriteByte(c)
|
|
if c == quote {
|
|
quote = 0
|
|
}
|
|
continue
|
|
}
|
|
switch c {
|
|
case '\'', '"':
|
|
quote = c
|
|
case '(', '[':
|
|
depth++
|
|
case ')', ']':
|
|
if depth > 0 {
|
|
depth--
|
|
}
|
|
case ',':
|
|
if depth == 0 {
|
|
out = append(out, cur.String())
|
|
cur.Reset()
|
|
continue
|
|
}
|
|
}
|
|
cur.WriteByte(c)
|
|
}
|
|
if strings.TrimSpace(cur.String()) != "" {
|
|
out = append(out, cur.String())
|
|
}
|
|
return out
|
|
}
|
|
|
|
// scanMatchingParen returns the index of the `)` that closes the `(`
|
|
// at openIdx, skipping single/double-quoted strings and `--` line
|
|
// comments. Returns -1 when unbalanced.
|
|
func scanMatchingParen(src []byte, openIdx int) int {
|
|
depth := 0
|
|
var quote byte
|
|
for i := openIdx; i < len(src); i++ {
|
|
c := src[i]
|
|
if quote != 0 {
|
|
if c == quote {
|
|
quote = 0
|
|
}
|
|
continue
|
|
}
|
|
switch c {
|
|
case '\'', '"':
|
|
quote = c
|
|
case '-':
|
|
if i+1 < len(src) && src[i+1] == '-' {
|
|
for i < len(src) && src[i] != '\n' {
|
|
i++
|
|
}
|
|
}
|
|
case '(':
|
|
depth++
|
|
case ')':
|
|
depth--
|
|
if depth == 0 {
|
|
return i
|
|
}
|
|
}
|
|
}
|
|
return -1
|
|
}
|
|
|
|
// lineCount returns the number of lines in src (at least 1).
|
|
func lineCount(src []byte) int {
|
|
n := 1
|
|
for _, b := range src {
|
|
if b == '\n' {
|
|
n++
|
|
}
|
|
}
|
|
return n
|
|
}
|
|
|
|
// truncateText caps s at max bytes, appending an ellipsis when cut.
|
|
func truncateText(s string, max int) string {
|
|
if len(s) <= max {
|
|
return s
|
|
}
|
|
return s[:max] + "…"
|
|
}
|