> ## Documentation Index > Fetch the complete documentation index at: https://docs.synq.io/llms.txt > Use this file to discover all available pages before exploring further. # Bi tool lineage ### bi\_tool\_lineage Find full example [here](https://github.com/getsynq/api/tree/main/examples/golang/bi_tool_lineage) #### main.go ```go theme={null} // Modelling a BI tool Coalesce Quality has no integration for, end to end. // // The estate this builds is four layers deep, and each hop uses a DIFFERENT // lineage mechanism, on purpose — picking the right one per hop is the whole // skill of modelling a new tool: // // warehouse tables (already in Coalesce Quality) // | SQL, resolved by warehouse address // v // BI models custom entities, SqlDefinition // | SQL, resolved by a BINDING — a model has no warehouse address // v // BI questions custom entities, SqlDefinition + references // | no SQL at all: declared column lineage // v // BI dashboard custom entity, ColumnLineage feature // | no columns: a plain relationship // v // BI subscription custom entity // // and off to one side, a BI alert custom entity, CheckCategory feature // attached with a check relationship // // Read metabase.go first — it is the tool's own metadata, and the mapping in // sync.go is easier to follow once you know what is being mapped. // // Everything written here is idempotent: re-running the program produces the // same estate, which is what lets it be a cron job rather than a migration. package main import ( "context" "crypto/tls" "fmt" "os" "time" entitiescustomv1grpc "buf.build/gen/go/getsynq/api/grpc/go/synq/entities/custom/v1/customv1grpc" lineagev1grpc "buf.build/gen/go/getsynq/api/grpc/go/synq/entities/lineage/v1/lineagev1grpc" "golang.org/x/oauth2/clientcredentials" "google.golang.org/grpc" "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/oauth" ) // Custom type ids are workspace-wide and yours to allocate: 1..1000, one number // per kind of thing your tool has. Pick a free band and keep it — an entity's // type is its id, so reusing a number silently reclassifies everything that had // it. checkTypeIdsAreFree below refuses to start if one of these is already in // use under another name. const ( typeModel = 40 typeQuestion = 41 typeDashboard = 42 typeSubscription = 43 typeAlert = 44 ) // groupID is the entity group every entity below is a member of. The group is // what makes a re-sync self-cleaning: content deleted in the BI tool simply // stops being sent, and the server deletes it. Nothing has to be remembered on // this side between runs. const groupID = "metabase" type config struct { endpoint string // The warehouse the BI tool queries. These name the DEFAULT execution // context of every query the tool runs, which is what lets an unqualified // table name in the SQL resolve to a real warehouse object. project string dataset string // Optional: the repository a serialized export of the BI content is // committed to, so Coalesce Quality can show its history. Leave unset to // skip that step. gitRepo string gitBranch string } func main() { ctx := context.Background() cfg := config{ // developer.synq.io (EU) is the default. The other deployments are // api.us.synq.io (US) and api.au.synq.io (AU); set QUALITY_API_ENDPOINT // to the one your workspace lives in. endpoint: env("QUALITY_API_ENDPOINT", "developer.synq.io"), project: env("BIGQUERY_PROJECT", "my-gcp-project"), dataset: env("BIGQUERY_DATASET", "analytics"), gitRepo: os.Getenv("BI_EXPORT_GIT_REPO"), gitBranch: env("BI_EXPORT_GIT_BRANCH", "main"), } clientID := env("QUALITY_CLIENT_ID", os.Getenv("SYNQ_CLIENT_ID")) clientSecret := env("QUALITY_CLIENT_SECRET", os.Getenv("SYNQ_CLIENT_SECRET")) if clientID == "" || clientSecret == "" { fmt.Println("set QUALITY_CLIENT_ID and QUALITY_CLIENT_SECRET (see README.md)") os.Exit(1) } oauthConfig := &clientcredentials.Config{ ClientID: clientID, ClientSecret: clientSecret, TokenURL: fmt.Sprintf("https://%s/oauth2/token", cfg.endpoint), } conn, err := grpc.NewClient( fmt.Sprintf("%s:443", cfg.endpoint), grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{InsecureSkipVerify: false})), grpc.WithPerRPCCredentials(oauth.TokenSource{TokenSource: oauthConfig.TokenSource(ctx)}), grpc.WithAuthority(cfg.endpoint), ) if err != nil { panic(err) } defer conn.Close() api := &clients{ types: entitiescustomv1grpc.NewTypesServiceClient(conn), entities: entitiescustomv1grpc.NewEntitiesServiceClient(conn), features: entitiescustomv1grpc.NewFeaturesServiceClient(conn), relationships: entitiescustomv1grpc.NewRelationshipsServiceClient(conn), checkRelationships: entitiescustomv1grpc.NewChecksRelationshipsServiceClient(conn), groups: entitiescustomv1grpc.NewGroupsServiceClient(conn), executions: entitiescustomv1grpc.NewEntityExecutionsServiceClient(conn), lineage: lineagev1grpc.NewLineageServiceClient(conn), } // The order below is not cosmetic. Three of these steps depend on an earlier // one having landed: // // - An entity carries a type id, so the types go first. // - A binding to a custom entity that does not exist is REFUSED, so every // entity exists before any SQL definition binds to one. // - A `SELECT *` over a bound model expands only if that model has declared // its columns, so schemas are written before the SQL that reads them. // // The last one is invisible when it is wrong: the table-level edge still // appears, only the column-level edges are missing. steps := []struct { name string run func(context.Context, *clients, config) error }{ {"check type ids are free", checkTypeIdsAreFree}, {"declare entity types", syncTypes}, {"declare entities", syncEntities}, {"declare schemas", syncSchemas}, {"declare model SQL (resolved by warehouse address)", syncModelSql}, {"declare question SQL (resolved by binding)", syncQuestionSql}, {"declare dashboard column lineage", syncDashboardColumnLineage}, {"attach the tool's own definitions as code", syncCode}, {"link the serialized export to git", syncGitFileReferences}, {"join the subscription to its dashboard", syncRelationships}, {"declare the alert as a check", syncAlerts}, {"reconcile the entity group", syncGroup}, {"report the last refresh of each dashboard", syncExecutions}, } for _, step := range steps { fmt.Printf("\n== %s\n", step.name) if err := step.run(ctx, api, cfg); err != nil { panic(fmt.Errorf("%s: %w", step.name, err)) } } // Lineage is computed asynchronously from what was just written: the SQL is // parsed, the bindings are resolved, and the graph is rebuilt. Reading it // back immediately usually finds it half-built, so this polls. fmt.Printf("\n== verify\n") verify(ctx, api, cfg, 2*time.Minute) } type clients struct { types entitiescustomv1grpc.TypesServiceClient entities entitiescustomv1grpc.EntitiesServiceClient features entitiescustomv1grpc.FeaturesServiceClient relationships entitiescustomv1grpc.RelationshipsServiceClient checkRelationships entitiescustomv1grpc.ChecksRelationshipsServiceClient groups entitiescustomv1grpc.GroupsServiceClient executions entitiescustomv1grpc.EntityExecutionsServiceClient lineage lineagev1grpc.LineageServiceClient } func env(name, fallback string) string { if v := os.Getenv(name); v != "" { return v } return fallback } ``` #### metabase.go ```go theme={null} package main // ----------------------------------------------------------------------------- // The BI tool's own metadata — the input half of the example. // ----------------------------------------------------------------------------- // // Metabase stands in for "a BI tool Coalesce Quality has no integration for". // It is a real product, and its object model is the one most BI tools have under // different names: // // Metabase Looker Sigma what it is // --------- ----------- ------------- -------------------------------- // model view data model a saved query others build on // question look workbook query a query, often built on a model // dashboard dashboard workbook a composition of question results // subscription schedule scheduled send a delivery of a dashboard // alert alert alert a condition watched on a question // // Rename the structs and the mapping in sync.go is unchanged, which is the point // of reading this file first: modelling a new tool is a mapping exercise, not an // API exercise. // // A real integration fetches all of this from the tool — for Metabase that is // GET /api/card, GET /api/dashboard/:id and GET /api/alert. It is declared inline // here so the example runs with nothing but Coalesce Quality credentials; the // `omni_types` example shows the fetch half against a live BI tool. // column is one field the tool exposes on a model, question or dashboard. type column struct { name string nativeType string desc string } // model is a saved, reusable query other content is built on. type model struct { id int name string description string collection string // sql is the query as the tool runs it against the warehouse. Table names in // it are real warehouse objects, so Coalesce Quality resolves them by address // and no binding is needed. sql string // sqlName is the name a DOWNSTREAM query writes when it reads this model. // Metabase itself compiles a nested question into a subquery and never emits // such a name, so the integration picks one — and picks it from the model's // stable id rather than its title, so renaming the model in Metabase does not // silently repoint every binding. A readable name is used here to keep the // example legible. sqlName string columns []column // definition is the tool's own object, carried across verbatim so a reviewer // can see what changed between two syncs without leaving Coalesce Quality. definition string } // question is a query, optionally built on one or more models. type question struct { id int name string description string collection string sql string // sourceModels maps a name this question's SQL writes to the model behind it. // Every name here needs a binding, because a model is not an object the // warehouse can address; the names NOT listed are ordinary warehouse tables // and resolve on their own. sourceModels map[string]int columns []column } // dashboard is a composition of question results. It runs no SQL of its own: a // field on a card is the question's column, carried through unchanged or // relabelled, which is why its lineage is declared rather than derived. type dashboard struct { id int name string description string collection string cards []card } // card is one field of one dashboard tile, and where its value comes from. // Exactly one of fromQuestion / fromWarehouseTable is set. type card struct { // field is the dashboard's own column name. field string // fromQuestion is the question id the value comes from, and fromColumn the // column on it. fromQuestion int // fromWarehouseTable is set instead when a tile reads the warehouse directly // — a filter widget populated from a column, say — rather than through a // question. fromWarehouseTable string fromColumn string } // subscription is a scheduled delivery of a dashboard. It has no columns of its // own, so it is joined to the dashboard by a plain relationship. type subscription struct { id int name string description string dashboard int schedule string } // alert is a condition watched on a question. It is a check: something that // passes or fails about another entity, rather than a stage data flows through. type alert struct { id int name string description string // question is the entity this alert checks. question int // kind is the tool's own name for what the alert does. It is reported as-is // and deliberately not mapped to a check category — see sync.go. kind string } // ----------------------------------------------------------------------------- // The sample content. // ----------------------------------------------------------------------------- // // Two models over the warehouse, two questions over the models, one dashboard // over the questions, one subscription and one alert. Small enough to follow in // the lineage graph, wide enough that every mechanism appears once. var models = []model{ { id: 31, name: "Order items enriched", description: "Order lines with product attributes and net revenue applied. The commercial team's starting point.", collection: "Commercial", sqlName: "order_items_enriched", sql: `SELECT oi.order_id AS order_id, oi.item_id AS item_id, oi.product_id AS product_id, p.name AS product_name, p.category AS product_category, oi.quantity AS quantity, oi.unit_price AS unit_price, oi.quantity * oi.unit_price * (1 - oi.discount) AS net_revenue FROM order_items AS oi JOIN products AS p ON p.id = oi.product_id`, columns: []column{ {"order_id", "INTEGER", "Order the line belongs to"}, {"item_id", "INTEGER", "Line identifier within the order"}, {"product_id", "INTEGER", "Product sold on this line"}, {"product_name", "STRING", "Product name at query time"}, {"product_category", "STRING", "Category the product belongs to"}, {"quantity", "INTEGER", "Units sold on this line"}, {"unit_price", "NUMERIC", "List price per unit"}, {"net_revenue", "NUMERIC", "quantity * unit_price after the line discount"}, }, definition: `{"id":31,"type":"model","name":"Order items enriched","collection":"Commercial","database_id":2}`, }, { id: 32, name: "Customer regions", description: "One row per customer with the region used for territory reporting.", collection: "Commercial", sqlName: "customer_regions_v", sql: `SELECT c.id AS customer_id, c.full_name AS customer_name, c.region AS region FROM customers AS c`, columns: []column{ {"customer_id", "INTEGER", "Customer surrogate key"}, {"customer_name", "STRING", "Customer display name"}, {"region", "STRING", "Sales region"}, }, definition: `{"id":32,"type":"model","name":"Customer regions","collection":"Commercial","database_id":2}`, }, } var questions = []question{ { id: 87, name: "Revenue by product category", description: "Net revenue and units, grouped by product category.", collection: "Commercial", // Reads one model and nothing else, so every table name in it is bound. sql: `SELECT m.product_category AS category, SUM(m.net_revenue) AS revenue, SUM(m.quantity) AS units FROM order_items_enriched AS m GROUP BY m.product_category`, sourceModels: map[string]int{"order_items_enriched": 31}, columns: []column{ {"category", "STRING", "Product category"}, {"revenue", "NUMERIC", "Net revenue for the category"}, {"units", "INTEGER", "Units sold in the category"}, }, }, { id: 88, name: "Customers by region", description: "Customer counts per region, with the contactable share.", collection: "Commercial", // Mixes both resolution paths in one definition: `customer_regions_v` is a // model and is bound, `customers` is a warehouse table and is not. sql: `SELECT r.region AS region, COUNT(DISTINCT r.customer_id) AS customers, COUNT(DISTINCT c.email) AS contactable_customers FROM customer_regions_v AS r JOIN customers AS c ON c.id = r.customer_id GROUP BY r.region`, sourceModels: map[string]int{"customer_regions_v": 32}, columns: []column{ {"region", "STRING", "Sales region"}, {"customers", "INTEGER", "Customers in the region"}, {"contactable_customers", "INTEGER", "Customers with an email address on file"}, }, }, } var dashboards = []dashboard{ { id: 9, name: "Commercial overview", description: "The weekly commercial read: revenue by category, customers by region.", collection: "Commercial", cards: []card{ {field: "revenue_by_category.category", fromQuestion: 87, fromColumn: "category"}, {field: "revenue_by_category.revenue", fromQuestion: 87, fromColumn: "revenue"}, {field: "revenue_by_category.units", fromQuestion: 87, fromColumn: "units"}, {field: "customers_by_region.region", fromQuestion: 88, fromColumn: "region"}, {field: "customers_by_region.customers", fromQuestion: 88, fromColumn: "customers"}, // A filter widget reading the warehouse directly, without a question in // between. A declared edge reaches a native entity exactly as it reaches // a custom one. {field: "filter.product_category", fromWarehouseTable: "products", fromColumn: "category"}, }, }, } var subscriptions = []subscription{ { id: 4, name: "Weekly commercial digest", description: "Sends the Commercial overview to the leadership list every Monday at 07:00.", dashboard: 9, schedule: "weekly, Monday 07:00 UTC", }, } var alerts = []alert{ { id: 12, name: "Revenue below weekly goal", description: "Fires when total net revenue for the week falls under the commercial goal.", question: 87, kind: "goal_alert", }, } ``` #### sync.go ```go theme={null} package main import ( "context" "fmt" "sort" "time" coordinatesv1 "buf.build/gen/go/getsynq/api/protocolbuffers/go/synq/entities/coordinates/v1" customfeaturesv1 "buf.build/gen/go/getsynq/api/protocolbuffers/go/synq/entities/custom/features/v1" customv1 "buf.build/gen/go/getsynq/api/protocolbuffers/go/synq/entities/custom/v1" entitiesv1 "buf.build/gen/go/getsynq/api/protocolbuffers/go/synq/entities/v1" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" ) // ----------------------------------------------------------------------------- // Identifiers // ----------------------------------------------------------------------------- // // A custom entity's id is yours, and it is the one thing that must never change: // every feature, relationship and declared edge hangs off it, and changing it // creates a second entity rather than renaming the first. So build it from what // is STABLE in the source system — the object's id — and never from its title. // The kind is in the id purely so a human reading a lineage graph can tell a // model from a question. func modelID(id int) *entitiesv1.Identifier { return customID(fmt.Sprintf("metabase::model::%d", id)) } func questionID(id int) *entitiesv1.Identifier { return customID(fmt.Sprintf("metabase::question::%d", id)) } func dashboardID(id int) *entitiesv1.Identifier { return customID(fmt.Sprintf("metabase::dashboard::%d", id)) } func subscriptionID(id int) *entitiesv1.Identifier { return customID(fmt.Sprintf("metabase::subscription::%d", id)) } func alertID(id int) *entitiesv1.Identifier { return customID(fmt.Sprintf("metabase::alert::%d", id)) } func customID(id string) *entitiesv1.Identifier { return &entitiesv1.Identifier{ Id: &entitiesv1.Identifier_Custom{Custom: &entitiesv1.CustomIdentifier{Id: id}}, } } // warehouseTable names an entity Coalesce Quality already has, from an // integration you did not write. Custom entities and native ones are the same // kind of thing to every API here: an edge, a binding or a declared column // works across the boundary without ceremony. func warehouseTable(cfg config, table string) *entitiesv1.Identifier { return &entitiesv1.Identifier{ Id: &entitiesv1.Identifier_BigqueryTable{ BigqueryTable: &entitiesv1.BigqueryTableIdentifier{ Project: cfg.project, Dataset: cfg.dataset, Table: table, }, }, } } // everyEntity is the full set this sync owns, in a stable order. The group step // sends exactly this, which is what makes deletion in the BI tool propagate. func everyEntity() []*entitiesv1.Identifier { var ids []*entitiesv1.Identifier for _, m := range models { ids = append(ids, modelID(m.id)) } for _, q := range questions { ids = append(ids, questionID(q.id)) } for _, d := range dashboards { ids = append(ids, dashboardID(d.id)) } for _, s := range subscriptions { ids = append(ids, subscriptionID(s.id)) } for _, a := range alerts { ids = append(ids, alertID(a.id)) } return ids } // ----------------------------------------------------------------------------- // Step 1 — types // ----------------------------------------------------------------------------- // checkTypeIdsAreFree refuses to start if one of this example's type ids is // already in use in the workspace under a different name. Type ids are // workspace-wide and there is no allocator, so the only protection against two // integrations picking the same number is to look before writing. func checkTypeIdsAreFree(ctx context.Context, api *clients, _ config) error { resp, err := api.types.ListTypes(ctx, &customv1.ListTypesRequest{}) if err != nil { return err } existing := map[int32]string{} for _, t := range resp.GetTypes() { existing[t.GetTypeId()] = t.GetName() } for _, want := range biTypes() { if name, taken := existing[want.GetTypeId()]; taken && name != want.GetName() { return fmt.Errorf( "type id %d is already used by %q: pick a different band in main.go", want.GetTypeId(), name, ) } } fmt.Printf(" %d type ids free or already ours\n", len(biTypes())) return nil } // biTypes is the tool's object model, one Type per kind of thing. // // Traits are the part worth getting right. A type declares how its entities // BEHAVE, and every entity of the type inherits it: is_model makes the BI // models rank and behave as transformation models do (the same way a dbt model // does), and is_bi_like marks the questions and dashboards as the leaves of the // graph a business reads. Get these wrong and the entities still appear, but // they sort, filter and rank as if they were something else. func biTypes() []*entitiesv1.Type { return []*entitiesv1.Type{ { TypeId: typeModel, Name: "Metabase Model", SvgIcon: iconLayers, Traits: &entitiesv1.TypeTraits{IsModel: proto.Bool(true)}, }, { TypeId: typeQuestion, Name: "Metabase Question", SvgIcon: iconChart, // A question transforms data AND is read by a person, so it is both. Traits: &entitiesv1.TypeTraits{IsModel: proto.Bool(true), IsBiLike: proto.Bool(true)}, }, { TypeId: typeDashboard, Name: "Metabase Dashboard", SvgIcon: iconTiles, Traits: &entitiesv1.TypeTraits{IsBiLike: proto.Bool(true)}, }, { TypeId: typeSubscription, Name: "Metabase Subscription", SvgIcon: iconEnvelope, Traits: &entitiesv1.TypeTraits{IsBiLike: proto.Bool(true)}, }, { TypeId: typeAlert, Name: "Metabase Alert", SvgIcon: iconWarning, // A check, not a stage data flows through. Traits: &entitiesv1.TypeTraits{IsTestType: proto.Bool(true)}, }, } } func syncTypes(ctx context.Context, api *clients, _ config) error { for _, t := range biTypes() { if _, err := api.types.UpsertType(ctx, &customv1.UpsertTypeRequest{Type: t}); err != nil { return err } fmt.Printf(" type %d %s\n", t.GetTypeId(), t.GetName()) } return nil } // ----------------------------------------------------------------------------- // Step 2 — entities // ----------------------------------------------------------------------------- // syncEntities creates every entity before anything points at one. Two later // steps depend on that: a SQL binding to a custom entity that does not exist is // rejected at the write, and an entity group whose members do not all exist is // not a set anyone can reconcile against. // // Annotations are the tool's own filing system carried across, so the catalog // can be filtered the way the BI tool is browsed. Keep them to labels a person // would pick out of a list; prose belongs in the description. func syncEntities(ctx context.Context, api *clients, cfg config) error { var entities []*entitiesv1.Entity for _, m := range models { entities = append(entities, &entitiesv1.Entity{ Id: modelID(m.id), TypeId: typeModel, Name: m.name, Description: m.description, Annotations: []*entitiesv1.Annotation{ {Name: "metabase.collection", Values: []string{m.collection}}, {Name: "metabase.kind", Values: []string{"model"}}, }, }) } for _, q := range questions { entities = append(entities, &entitiesv1.Entity{ Id: questionID(q.id), TypeId: typeQuestion, Name: q.name, Description: q.description, Annotations: []*entitiesv1.Annotation{ {Name: "metabase.collection", Values: []string{q.collection}}, {Name: "metabase.kind", Values: []string{"question"}}, }, }) } for _, d := range dashboards { entities = append(entities, &entitiesv1.Entity{ Id: dashboardID(d.id), TypeId: typeDashboard, Name: d.name, Description: d.description, Annotations: []*entitiesv1.Annotation{ {Name: "metabase.collection", Values: []string{d.collection}}, {Name: "metabase.kind", Values: []string{"dashboard"}}, }, }) } for _, s := range subscriptions { entities = append(entities, &entitiesv1.Entity{ Id: subscriptionID(s.id), TypeId: typeSubscription, Name: s.name, Description: fmt.Sprintf("%s\n\nSchedule: %s.", s.description, s.schedule), Annotations: []*entitiesv1.Annotation{ {Name: "metabase.kind", Values: []string{"subscription"}}, }, }) } for _, a := range alerts { entities = append(entities, &entitiesv1.Entity{ Id: alertID(a.id), TypeId: typeAlert, Name: a.name, Description: a.description, Annotations: []*entitiesv1.Annotation{ {Name: "metabase.kind", Values: []string{"alert"}}, }, }) } for _, e := range entities { if _, err := api.entities.UpsertEntity(ctx, &customv1.UpsertEntityRequest{Entity: e}); err != nil { return err } fmt.Printf(" %s %s\n", e.GetId().GetCustom().GetId(), e.GetName()) } _ = cfg return nil } // ----------------------------------------------------------------------------- // Step 3 — schemas // ----------------------------------------------------------------------------- // syncSchemas declares what columns each entity has. // // This is not decoration, and it is the step most easily skipped. A declared // schema is what lets the next entity down read the columns of this one: a // `SELECT *` over a bound model expands only over a model whose columns are // known, and a declared column edge only shows against a column the entity // actually has. A dashboard with no schema still appears in table-level // lineage, and silently carries no column lineage at all. func syncSchemas(ctx context.Context, api *clients, _ config) error { put := func(id *entitiesv1.Identifier, cols []column) error { schema := &customfeaturesv1.Schema{ StateAt: timestamppb.Now(), Columns: make([]*entitiesv1.SchemaColumn, 0, len(cols)), } for i, c := range cols { schema.Columns = append(schema.Columns, &entitiesv1.SchemaColumn{ Name: c.name, NativeType: c.nativeType, Description: c.desc, OrdinalPosition: int32(i + 1), }) } _, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{ Feature: &customv1.Feature{ EntityId: id, // One schema feature per entity, so the id is a constant. Never // generate it — a fresh id on every run creates a new feature // each time instead of replacing the previous one. FeatureId: "schema", Feature: &customv1.Feature_Schema{Schema: schema}, }, }) if err != nil { return err } fmt.Printf(" %s %d columns\n", id.GetCustom().GetId(), len(cols)) return nil } for _, m := range models { if err := put(modelID(m.id), m.columns); err != nil { return err } } for _, q := range questions { if err := put(questionID(q.id), q.columns); err != nil { return err } } for _, d := range dashboards { cols := make([]column, 0, len(d.cards)) for _, c := range d.cards { cols = append(cols, column{name: c.field, nativeType: "", desc: ""}) } if err := put(dashboardID(d.id), cols); err != nil { return err } } return nil } // ----------------------------------------------------------------------------- // Step 4 — model SQL, resolved by warehouse address // ----------------------------------------------------------------------------- // warehouseContext is the default execution context of every query the BI tool // runs: the database and schema an unqualified table name in its SQL resolves // against. It is the tool's connection settings, expressed the way Coalesce // Quality addresses a warehouse object. // // object_name is deliberately left empty. It is for a definition that // MATERIALIZES something — `CREATE TABLE x AS SELECT ...` — and nothing in a BI // tool does; a model is a saved query, not a table. func warehouseContext(cfg config) *coordinatesv1.DatabaseContext { return &coordinatesv1.DatabaseContext{ // BigQuery has no instance above the project, so instance_name stays // empty. On Snowflake this would be the account, on Databricks the // workspace URL. DatabaseName: cfg.project, SchemaName: cfg.dataset, } } // syncModelSql declares the models' SQL. Every table name in it is a real // warehouse object, so it resolves by address and no binding is needed — this // is the plain case, and the one to reach for whenever it fits. // // Coalesce Quality parses the SQL and derives BOTH grains from it: the model // appears downstream of `order_items` and `products`, and `net_revenue` traces // back to `order_items.quantity`, `order_items.unit_price` and // `order_items.discount` without any of that being stated. That is the whole // argument for giving SQL rather than declaring edges: the lineage follows the // SQL when the SQL changes. func syncModelSql(ctx context.Context, api *clients, cfg config) error { for _, m := range models { _, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{ Feature: &customv1.Feature{ EntityId: modelID(m.id), FeatureId: "sql", Feature: &customv1.Feature_SqlDefinition{ SqlDefinition: &customfeaturesv1.SqlDefinition{ StateAt: timestamppb.Now(), Dialect: entitiesv1.SqlDialect_SQL_DIALECT_BIGQUERY, Sql: m.sql, DatabaseContext: warehouseContext(cfg), }, }, }, }) if err != nil { return err } fmt.Printf(" %s %d chars of SQL, no bindings\n", modelID(m.id).GetCustom().GetId(), len(m.sql)) } return nil } // ----------------------------------------------------------------------------- // Step 5 — question SQL, resolved by a binding // ----------------------------------------------------------------------------- // syncQuestionSql declares the questions' SQL, and this is the headline. // // A question reads a MODEL, and a model is not something the warehouse can // address: it has no database, schema and table for a lookup to land on. Left // alone, `FROM order_items_enriched` resolves to nothing, is dropped, and the // question ends up with no upstream at all — with no error anywhere, and an // empty result indistinguishable from a query that genuinely reads nothing. // // `references` is how the definition says what such a name means. Each binding // pairs a name AS THE SQL WRITES IT with the entity it stands for, and the // parse then derives both grains from the SQL as usual: the table edge, and // `revenue` tracing through `order_items_enriched.net_revenue` all the way down // to the warehouse columns the model computed it from. // // Three things worth knowing: // // - A binding wins over the warehouse. If a real table happened to share the // name, the binding is still what resolves — which is what makes it usable // as a manual override, not just a fallback. // - Empty parts of a binding default from the definition's own // database_context, exactly as an unqualified name in the SQL does. So a // bare object_name matches both `FROM order_items_enriched` and // `FROM ..order_items_enriched`. // - Names that ARE warehouse objects need no binding. Question 88 below reads // a model and a raw table in one query, and only the model is listed. func syncQuestionSql(ctx context.Context, api *clients, cfg config) error { byID := map[int]model{} for _, m := range models { byID[m.id] = m } for _, q := range questions { // Sorted so two runs send byte-identical requests, which is what keeps a // re-sync from being mistaken for a change. names := make([]string, 0, len(q.sourceModels)) for name := range q.sourceModels { names = append(names, name) } sort.Strings(names) refs := make([]*customfeaturesv1.SqlTableReference, 0, len(names)) for _, name := range names { m, ok := byID[q.sourceModels[name]] if !ok { return fmt.Errorf("question %d binds %q to unknown model %d", q.id, name, q.sourceModels[name]) } refs = append(refs, &customfeaturesv1.SqlTableReference{ // The name exactly as the SQL writes it. Matched // case-insensitively; leave database_name and schema_name empty // to default them from database_context. ObjectName: name, Entity: modelID(m.id), }) } _, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{ Feature: &customv1.Feature{ EntityId: questionID(q.id), FeatureId: "sql", Feature: &customv1.Feature_SqlDefinition{ SqlDefinition: &customfeaturesv1.SqlDefinition{ StateAt: timestamppb.Now(), Dialect: entitiesv1.SqlDialect_SQL_DIALECT_BIGQUERY, Sql: q.sql, DatabaseContext: warehouseContext(cfg), // The bindings are part of the definition and replace with // it: one dropped from a later write is gone, and the // lineage it produced is withdrawn. References: refs, }, }, }, }) if err != nil { return err } for _, r := range refs { fmt.Printf(" %s %q -> %s\n", questionID(q.id).GetCustom().GetId(), r.GetObjectName(), r.GetEntity().GetCustom().GetId()) } } return nil } // ----------------------------------------------------------------------------- // Step 6 — dashboard column lineage, declared // ----------------------------------------------------------------------------- // syncDashboardColumnLineage states the dashboard's column lineage outright. // // A dashboard runs no SQL: a tile takes a question's result and shows it, // possibly under a different label. There is nothing for a parser to read, so // the answer is declared instead of derived. That is the rule of thumb for the // whole exercise — SqlDefinition when the component has SQL, ColumnLineage when // it does not. // // The declaration is COMPLETE and replaces the previous one whole: an edge left // out of a write is withdrawn, and re-sending the same set changes nothing. // That is what makes it safe to regenerate from the tool's metadata on every // run rather than diffing against what was sent last time. // // Declaring a column edge also declares the table edge it implies, so the // dashboard appears downstream of each question here without a relationship // being written too. func syncDashboardColumnLineage(ctx context.Context, api *clients, cfg config) error { for _, d := range dashboards { edges := make([]*customfeaturesv1.ColumnEdge, 0, len(d.cards)) for _, c := range d.cards { upstream := questionID(c.fromQuestion) if c.fromWarehouseTable != "" { // An upstream on a connected platform works exactly the same way, // and it does not have to exist yet: the edge is remembered and // appears once the entity is next ingested. upstream = warehouseTable(cfg, c.fromWarehouseTable) } edges = append(edges, &customfeaturesv1.ColumnEdge{ Upstream: upstream, UpstreamColumn: c.fromColumn, // The two sides are named independently, so they need not match — // here the dashboard prefixes each field with the tile it sits on. Column: c.field, }) } _, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{ Feature: &customv1.Feature{ EntityId: dashboardID(d.id), // Only one column-lineage feature is allowed per entity, so a // stable id is how it is edited. FeatureId: "column-lineage", Feature: &customv1.Feature_ColumnLineage{ ColumnLineage: &customfeaturesv1.ColumnLineage{ StateAt: timestamppb.Now(), Edges: edges, }, }, }, }) if err != nil { return err } fmt.Printf(" %s %d column edges declared\n", dashboardID(d.id).GetCustom().GetId(), len(edges)) } return nil } // ----------------------------------------------------------------------------- // Step 7 — the tool's own definitions, as code // ----------------------------------------------------------------------------- // syncCode attaches each object's definition as it comes out of the BI tool. // Coalesce Quality shows it beside the entity and tracks what changed between // versions, which is how a question that quietly started filtering differently // gets noticed. // // Code and SqlDefinition are not alternatives: SqlDefinition is parsed and // produces lineage, Code is displayed and produces history. The models below // have both — the SQL for the graph, the card definition for the diff. // // Unlike schema and SQL, an entity may carry several code features, so the // feature id is the file name it stands for. func syncCode(ctx context.Context, api *clients, _ config) error { for _, m := range models { _, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{ Feature: &customv1.Feature{ EntityId: modelID(m.id), FeatureId: fmt.Sprintf("card_%d.json", m.id), Feature: &customv1.Feature_Code{ Code: &customfeaturesv1.Code{ Name: fmt.Sprintf("card_%d.json", m.id), CodeType: entitiesv1.CodeType_CODE_TYPE_JSON, Content: m.definition, }, }, }, }) if err != nil { return err } fmt.Printf(" %s card_%d.json\n", modelID(m.id).GetCustom().GetId(), m.id) } return nil } // ----------------------------------------------------------------------------- // Step 8 — the serialized export in git // ----------------------------------------------------------------------------- // syncGitFileReferences points each dashboard at the file its serialized export // lives in, when the BI content is version-controlled (Metabase calls this // serialization; most tools have an equivalent). Coalesce Quality then reads the // file's history from the repository, so "what changed, and who changed it" is // answered from the commit rather than from the tool's audit log. // // Skipped unless BI_EXPORT_GIT_REPO is set, because a reference to a repository // that does not exist is worse than no reference. func syncGitFileReferences(ctx context.Context, api *clients, cfg config) error { if cfg.gitRepo == "" { fmt.Println(" skipped: BI_EXPORT_GIT_REPO is not set") return nil } for _, d := range dashboards { path := fmt.Sprintf("collections/%s/dashboards/%d.yaml", d.collection, d.id) _, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{ Feature: &customv1.Feature{ EntityId: dashboardID(d.id), FeatureId: path, Feature: &customv1.Feature_GitFileReference{ GitFileReference: &customfeaturesv1.GitFileReference{ RepositoryUrl: cfg.gitRepo, BranchName: cfg.gitBranch, FilePath: path, }, }, }, }) if err != nil { return err } fmt.Printf(" %s %s\n", dashboardID(d.id).GetCustom().GetId(), path) } return nil } // ----------------------------------------------------------------------------- // Step 9 — relationships with no columns // ----------------------------------------------------------------------------- // syncRelationships joins each subscription to the dashboard it sends. // // A subscription has no columns of its own — it is a delivery, not a // transformation — so there is nothing to state at column grain and a plain // relationship is the right shape. Reach for one whenever the edge is real but // the column mapping is not: a reverse-ETL job, an export, a downstream service. // // The response reports what each write DID. Read it rather than assuming: an // edge is held between the two entities the endpoints RESOLVE to, and one entity // accepts several spellings, so two relationships that look different can be the // same edge. func syncRelationships(ctx context.Context, api *clients, _ config) error { var rels []*customv1.Relationship for _, s := range subscriptions { rels = append(rels, &customv1.Relationship{ Upstream: dashboardID(s.dashboard), Downstream: subscriptionID(s.id), }) } resp, err := api.relationships.UpsertRelationships(ctx, &customv1.UpsertRelationshipsRequest{ Relationships: rels, }) if err != nil { return err } for _, r := range resp.GetResults() { fmt.Printf(" %s -> %s %s\n", r.GetRelationship().GetUpstream().GetEntityId(), r.GetRelationship().GetDownstream().GetEntityId(), r.GetOutcome(), ) } return nil } // ----------------------------------------------------------------------------- // Step 10 — the alert as a check // ----------------------------------------------------------------------------- // syncAlerts models the BI tool's alert as a CHECK on the question it watches, // rather than as another node data flows through. A check relationship is a // different edge from a lineage one: it says "this validates that", so the // alert's state rolls up to the question's health instead of extending the // graph. // // The categories are deliberately left unset. `category` (what kind of check // this is, mechanically) and `governance_category` (what the check is for) are // resolved by the workspace's own categorisation rules, and a value sent here // OUTRANKS those rules. Deriving one from `kind` — which is the obvious thing to // do, and wrong — would silently switch off every rule the workspace wrote, with // nothing in the product explaining why. Send what the tool actually says // (`package` and `kind`) and let the rules decide the rest. func syncAlerts(ctx context.Context, api *clients, _ config) error { for _, a := range alerts { _, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{ Feature: &customv1.Feature{ EntityId: alertID(a.id), FeatureId: "check", Feature: &customv1.Feature_CheckCategory{ CheckCategory: &customfeaturesv1.CheckCategory{ Package: "metabase", Kind: a.kind, }, }, }, }) if err != nil { return err } resp, err := api.checkRelationships.UpsertCheckRelationships(ctx, &customv1.UpsertCheckRelationshipsRequest{ CheckRelationships: []*customv1.CheckRelationship{{ Check: alertID(a.id), Checked: questionID(a.question), }}, }) if err != nil { return err } for _, r := range resp.GetResults() { fmt.Printf(" %s checks %s %s\n", r.GetCheckRelationship().GetCheck().GetEntityId(), r.GetCheckRelationship().GetChecked().GetEntityId(), r.GetOutcome(), ) } } return nil } // ----------------------------------------------------------------------------- // Step 11 — the group, which is what makes deletion work // ----------------------------------------------------------------------------- // syncGroup sends the complete set of entities this integration owns. // // The server holds the previous set, diffs it, and deletes what is no longer // there. That is the whole reason to use a group: without one, an integration // has to remember on its own side what it created last time in order to clean up // after a dashboard someone deleted in the BI tool — and any state it keeps for // that will eventually disagree with reality. // // Send the FULL set every run. A partial send is not an update, it is a // deletion of everything omitted. func syncGroup(ctx context.Context, api *clients, _ config) error { ids := everyEntity() resp, err := api.groups.UpsertEntitiesGroup(ctx, &customv1.UpsertEntitiesGroupRequest{ Group: &customv1.Group{GroupId: groupID, EntityIds: ids}, }) if err != nil { return err } fmt.Printf(" group %q now holds %d entities\n", groupID, len(ids)) for _, deleted := range resp.GetDeletedIds() { fmt.Printf(" deleted (gone from the BI tool): %s\n", deleted.GetCustom().GetId()) } return nil } // ----------------------------------------------------------------------------- // Step 12 — executions // ----------------------------------------------------------------------------- // syncExecutions reports each dashboard's last refresh, which is what gives a // custom entity a status: fresh, stale, failing. Without one it is a node in a // graph with nothing known about its health. // // Report what the tool tells you and nothing more. `created_at` must be in the // past, so a refresh that has not happened yet is simply not reported. func syncExecutions(ctx context.Context, api *clients, _ config) error { // A real integration takes this from the tool's own run history. finished := time.Now().Add(-15 * time.Minute) started := finished.Add(-42 * time.Second) for _, d := range dashboards { _, err := api.executions.UpsertExecution(ctx, &customv1.UpsertExecutionRequest{ Execution: &customv1.Execution{ Id: dashboardID(d.id), Status: customv1.ExecutionStatus_EXECUTION_STATUS_OK, Message: "All cards refreshed.", CreatedAt: timestamppb.New(finished), StartedAt: timestamppb.New(started), FinishedAt: timestamppb.New(finished), }, }) if err != nil { return err } fmt.Printf(" %s refreshed %s\n", dashboardID(d.id).GetCustom().GetId(), finished.Format(time.RFC3339)) } return nil } // ----------------------------------------------------------------------------- // Icons // ----------------------------------------------------------------------------- // // Any SVG works. These are plain geometry so the example carries no licence // question; swap in the BI tool's own mark when you adapt it. var ( iconLayers = []byte(``) iconChart = []byte(``) iconTiles = []byte(``) iconEnvelope = []byte(``) iconWarning = []byte(``) ) ``` #### verify.go ```go theme={null} package main import ( "context" "fmt" "strings" "time" lineagev1 "buf.build/gen/go/getsynq/api/protocolbuffers/go/synq/entities/lineage/v1" entitiesv1 "buf.build/gen/go/getsynq/api/protocolbuffers/go/synq/entities/v1" ) // verify reads the graph back. // // None of this is needed to write the estate — it is here because the way this // goes wrong is silent. A missing binding, a schema that was never declared, a // column edge whose table edge is absent: each of them leaves a valid-looking // write and an empty result. Reading the lineage back after a sync is the only // thing that distinguishes "no upstreams" from "upstreams that did not resolve". // // Lineage is computed asynchronously from what was written: the SQL is parsed, // bindings are resolved, the graph is rebuilt. So this polls rather than asking // once. func verify(ctx context.Context, api *clients, cfg config, budget time.Duration) { dash := dashboardID(dashboards[0].id) // The table-level chain. Expect five hops: dashboard <- question <- model <- // warehouse tables, plus the filter's direct warehouse edge. wantTableNodes := 7 fmt.Printf("\n-- table lineage upstream of %s\n", dash.GetCustom().GetId()) table := poll(ctx, budget, func() (*lineagev1.Lineage, bool) { lin := getLineage(ctx, api, &lineagev1.GetLineageStartPoint{ From: &lineagev1.GetLineageStartPoint_Entities{ Entities: &lineagev1.EntitiesStartPoint{Entities: []*entitiesv1.Identifier{dash}}, }, }) return lin, lin != nil && len(lin.GetNodes()) >= wantTableNodes }) printTableLineage(table) // The column-level chain, which is the claim the whole example is making: a // field on a dashboard that runs no SQL traces through a question and a // model, neither of which the warehouse can address, to the physical column // the number came from. col := "revenue_by_category.revenue" fmt.Printf("\n-- column lineage upstream of %s.%s\n", dash.GetCustom().GetId(), col) cll := poll(ctx, budget, func() (*lineagev1.Lineage, bool) { lin := getLineage(ctx, api, &lineagev1.GetLineageStartPoint{ From: &lineagev1.GetLineageStartPoint_EntityColumns{ EntityColumns: &lineagev1.EntityColumnsStartPoint{ Id: dash, ColumnNames: []string{col}, }, }, }) return lin, lin != nil && len(lin.GetColumnDependencies()) > 0 }) printColumnLineage(cll, cfg) } func getLineage(ctx context.Context, api *clients, from *lineagev1.GetLineageStartPoint) *lineagev1.Lineage { depth := int32(10) resp, err := api.lineage.GetLineage(ctx, &lineagev1.GetLineageRequest{ LineageDirection: lineagev1.LineageDirection_LINEAGE_DIRECTION_UPSTREAM, StartPoint: from, MaxDepth: &depth, }) if err != nil { fmt.Printf(" (lineage read failed: %v)\n", err) return nil } return resp.GetLineage() } func poll(ctx context.Context, budget time.Duration, once func() (*lineagev1.Lineage, bool)) *lineagev1.Lineage { deadline := time.Now().Add(budget) for { lin, done := once() if done { return lin } if time.Now().After(deadline) { fmt.Printf(" (gave up after %s — the graph may still be building)\n", budget) return lin } select { case <-ctx.Done(): return lin case <-time.After(10 * time.Second): } } } func printTableLineage(lin *lineagev1.Lineage) { if lin == nil { return } for i, n := range lin.GetNodes() { fmt.Printf(" [%d] %-14s %s\n", i, strings.TrimPrefix(n.GetPosition().String(), "NODE_POSITION_"), nodeName(n)) } for _, d := range lin.GetNodeDependencies() { fmt.Printf(" %s -> %s\n", nodeName(lin.GetNodes()[d.GetSourceNodeIdx()]), nodeName(lin.GetNodes()[d.GetTargetNodeIdx()]), ) } } func printColumnLineage(lin *lineagev1.Lineage, _ config) { if lin == nil { return } for _, n := range lin.GetNodes() { if state := n.GetCllDetails().GetCllState(); state != lineagev1.CllState_CLL_STATE_OK && state != lineagev1.CllState_CLL_STATE_UNSPECIFIED { // Worth surfacing: RESOLUTION_FAILED on a node is usually a name the // definition should have bound and did not. fmt.Printf(" %s: %s %v\n", nodeName(n), strings.TrimPrefix(state.String(), "CLL_STATE_"), n.GetCllDetails().GetCllMessages()) } } for _, d := range lin.GetColumnDependencies() { fmt.Printf(" %s.%s -> %s.%s\n", nodeName(lin.GetNodes()[d.GetSourceNodeIdx()]), d.GetSourceNodeColumnId(), nodeName(lin.GetNodes()[d.GetTargetNodeIdx()]), d.GetTargetNodeColumnId(), ) } } // nodeName prefers the identifier shape that says what the node IS. A node // carries every identifier that resolves to it, so a warehouse table reached // through a custom entity's binding may list several. func nodeName(n *lineagev1.LineageNode) string { for _, id := range n.GetIds() { if c := id.GetCustom(); c != nil { return c.GetId() } if b := id.GetBigqueryTable(); b != nil { return fmt.Sprintf("%s.%s.%s", b.GetProject(), b.GetDataset(), b.GetTable()) } } if len(n.GetIds()) > 0 { return n.GetIds()[0].GetEntityId() } return "?" } ```