Documentation
¶
Overview ¶
Package catalogimport is the one-time import of a deployment's DataHub into the platform catalog, and the rewrite of the urn:li: references platform tables hold into the catalog's mcp: references (#1923).
The import reads every dataset DataHub holds, the records it has marked removed among them, and every glossary node, glossary term, tag, domain and data product. A dataset's curation lands on the catalog dataset with the same connection, catalog, schema and table, which a scan must have cataloged first. Curation already written in the platform catalog by anyone else is kept: the import fills, it does not overwrite. Running it again writes only what changed in DataHub since.
Index ¶
- Constants
- Variables
- type Catalog
- type Client
- type ContextDocument
- type DataHub
- type DatasetCounts
- type DatasetHead
- type DatasetRecord
- type Definition
- type GraphQL
- type Importer
- type Mapping
- type Options
- type Report
- type Resolver
- type Rewrite
- type RewriteReport
- type Rewriter
- type Run
- type Runner
- func (rn *Runner) Configure(src *SourceConfig)
- func (rn *Runner) Configured() string
- func (rn *Runner) ImportOnUpgrade(ctx context.Context)
- func (rn *Runner) Runs(ctx context.Context, limit int) ([]Run, error)
- func (rn *Runner) Start(ctx context.Context, url, token, by string) (Run, error)
- func (rn *Runner) StartConfigured(ctx context.Context, by string) (Run, error)
- func (rn *Runner) Stop()
- type Skip
- type Source
- type SourceConfig
- type SourceFactory
- type Store
Constants ¶
const ( ReasonNotCataloged = "no connection's catalog holds a dataset with this path; " + "scan the connection that sees it, then import again" ReasonPlatform = "its platform is not a connection kind the catalog scans" ReasonTableau = "Tableau entities are not imported" ReasonUnmapped = "the import has no catalog entity for this URN" ReasonUnsupported = "this kind of DataHub entity has no catalog counterpart" ReasonKept = "written in the platform catalog since, so the catalog's value is kept" ReasonNoColumn = "the catalog dataset has no column with this name" ReasonUnreadable = "DataHub could not read it" )
The reasons a dataset or reference is not carried over.
const LockKey int64 = 4713210202
LockKey is the advisory lock a run holds, so two administrators, or two replicas, never import at once.
const Principal = "datahub-import"
Principal is who the import's catalog writes are recorded as.
const StartedByUpgrade = "platform (upgrade from DataHub)"
StartedByUpgrade is who a run the platform started on upgrade is recorded as.
Variables ¶
var ErrNoSource = errors.New("the configuration names no DataHub (toolkits.datahub); give its url and token")
ErrNoSource reports a start that names no DataHub when the configuration names none either.
var ErrRunning = errors.New("an import from DataHub is already running")
ErrRunning reports an import already under way.
Functions ¶
This section is empty.
Types ¶
type Catalog ¶
type Catalog interface {
GetDataset(ctx context.Context, ref catalog.Ref) (*catalog.Dataset, error)
GetDefinition(ctx context.Context, ref catalog.Ref) (*catalog.Definition, error)
PutDefinition(ctx context.Context, in catalog.DefinitionInput) (*catalog.Definition, error)
History(ctx context.Context, ref catalog.Ref, name catalog.AspectName) ([]catalog.Change, error)
WriteAspects(ctx context.Context, edits []catalog.AspectEdit, by string) ([]catalog.Change, error)
AddEdges(ctx context.Context, edges []catalog.Edge) error
DatasetsByPath(ctx context.Context, connectionKind, path string) ([]string, error)
}
Catalog is what the import needs from the catalog store.
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client reads DataHub's GraphQL API with a personal access token.
func NewClient ¶
NewClient reads the DataHub at address, the GMS address with or without /api/graphql, with token.
func (*Client) Execute ¶
func (c *Client) Execute(ctx context.Context, query string, variables map[string]any, result any) error
Execute runs one GraphQL query and decodes its data into result. A GraphQL error, an answer with no data, and any status but 200 are errors.
func (*Client) GetContextDocuments ¶
GetContextDocuments reads the context documents DataHub keeps on a dataset, at most maxContextDocuments of them.
type ContextDocument ¶
ContextDocument is a markdown note DataHub keeps on an entity.
type DataHub ¶
type DataHub struct {
// contains filtered or unexported fields
}
DataHub reads what the import carries over from a DataHub instance.
func (*DataHub) DatasetHeads ¶
func (d *DataHub) DatasetHeads(ctx context.Context) ([]DatasetHead, error)
DatasetHeads lists every dataset, removed ones included, saying which are removed.
func (*DataHub) Definitions ¶
func (d *DataHub) Definitions(ctx context.Context) ([]Definition, error)
Definitions reads every glossary node, glossary term, tag, domain and data product that is not marked removed, nodes first.
type DatasetCounts ¶
type DatasetCounts struct {
// Read is every dataset record DataHub listed, removed ones included.
Read int `json:"read"`
// Imported is the catalog datasets the import wrote curation onto.
Imported int `json:"imported"`
// FromRemoved is the catalog datasets that took curation from a record
// DataHub has marked removed.
FromRemoved int `json:"from_removed"`
Skipped int `json:"skipped"`
}
DatasetCounts counts the datasets the import read and what became of them.
type DatasetHead ¶
DatasetHead is a dataset DataHub lists: its URN and whether DataHub has marked it removed.
type DatasetRecord ¶
type DatasetRecord struct {
URN string
Removed bool
Aspects catalog.Aspects
// Columns are the curation of each top-level field, by field name.
Columns map[string]catalog.Aspects
// Upstreams are the URNs of the datasets DataHub records this one is
// derived from.
Upstreams []string
}
DatasetRecord is one DataHub dataset's curation in the catalog's shapes. Glossary terms and the domain are still DataHub URNs; the import maps them.
type Definition ¶
type Definition struct {
URN string
Kind catalog.Kind
Name string
// Parent is the URN of the glossary node a term or node sits under.
Parent string
Aspects catalog.Aspects
}
Definition is a DataHub glossary node, glossary term, tag, domain or data product.
type GraphQL ¶
type GraphQL interface {
Execute(ctx context.Context, query string, variables map[string]any, result any) error
GetContextDocuments(ctx context.Context, urn string) ([]ContextDocument, error)
}
GraphQL is the part of the DataHub client the import reads through: raw GraphQL, and the context documents the client already reads.
type Importer ¶
type Importer struct {
// contains filtered or unexported fields
}
Importer carries a DataHub's curation into the platform catalog.
type Mapping ¶
type Mapping interface {
Map(ctx context.Context) (map[string]string, error)
Record(ctx context.Context, urn, reference string) error
}
Mapping is the import's record of what each DataHub URN became.
type Options ¶
type Options struct {
// Platform is the DataHub platform the deployment's Trino datasets were
// recorded under when it is not "trino".
Platform string
// CatalogMapping maps a query-engine catalog name to the name DataHub
// knows it by; the import reads it backwards.
CatalogMapping map[string]string
}
Options are the deployment's URN settings the import reads DataHub's dataset names by (semantic.urn_mapping).
type Report ¶
type Report struct {
// Definitions counts the definitions imported, by kind.
Definitions map[string]int `json:"definitions"`
Datasets DatasetCounts `json:"datasets"`
// AspectsWritten counts the aspects the import wrote; AspectsUnchanged
// those that already held DataHub's value; AspectsKept those it left
// alone because something other than the import wrote them in the
// platform catalog.
AspectsWritten int `json:"aspects_written"`
AspectsUnchanged int `json:"aspects_unchanged"`
AspectsKept int `json:"aspects_kept"`
LineageEdges int `json:"lineage_edges"`
// Skipped lists what was not imported and why, at most maxListed of
// them; SkippedByReason counts every one.
Skipped []Skip `json:"skipped"`
SkippedByReason map[string]int `json:"skipped_by_reason"`
// Rewrite is what the rewrite of stored references did.
Rewrite *RewriteReport `json:"rewrite,omitempty"`
}
Report is what one run of the import did.
type Resolver ¶
Resolver turns a DataHub URN into the catalog reference it became, or "" when there is none.
type RewriteReport ¶
type RewriteReport struct {
// Rewritten counts the rows changed, by table and column.
Rewritten map[string]int `json:"rewritten"`
// Unmapped counts the references left as they were, because the import
// has no catalog entity for them; each is listed in the report's skips
// with where it was found.
Unmapped int `json:"unmapped"`
// IndexRowsDropped counts the rows of the dataset search index keyed by a
// DataHub URN. The index reads the platform catalog; it re-reads the
// datasets those rows named under their catalog references.
IndexRowsDropped int `json:"index_rows_dropped"`
}
RewriteReport is what the rewrite of stored references did.
type Rewriter ¶
type Rewriter struct {
// contains filtered or unexported fields
}
Rewriter rewrites the DataHub URNs platform tables hold into catalog references.
type Run ¶
type Run struct {
ID string `json:"id"`
SourceURL string `json:"source_url"`
StartedBy string `json:"started_by"`
StartedAt time.Time `json:"started_at"`
FinishedAt *time.Time `json:"finished_at,omitempty"`
// Running is true while the run holds the import's lock; a run with no
// finished_at that does not was cut off when its replica stopped.
Running bool `json:"running"`
Report *Report `json:"report,omitempty"`
Error string `json:"error,omitempty"`
}
Run is one run of the import as every replica reports it.
type Runner ¶
type Runner struct {
// contains filtered or unexported fields
}
Runner starts imports in the background, one at a time across replicas, and records each run.
func NewRunner ¶
NewRunner builds a runner that imports with importer, rewrites with rw and reads the DataHub open returns.
func (*Runner) Configure ¶
func (rn *Runner) Configure(src *SourceConfig)
Configure names the DataHub the configuration names: the one an upgrade imports from, and the one a start that names none reads.
func (*Runner) Configured ¶
Configured is the address of the DataHub the configuration names, or "".
func (*Runner) ImportOnUpgrade ¶
ImportOnUpgrade begins an import from the DataHub the configuration names unless a run against it has finished without error. It is called after each catalog scan, so the import reads a catalog that holds the deployment's datasets, and a failed import is tried again after the next scan. Another replica's run under way is left to finish.
func (*Runner) Start ¶
Start begins an import from the DataHub at url and returns its run. It answers ErrRunning while another run holds the lock, and an error naming the problem when the DataHub cannot be opened.
func (*Runner) StartConfigured ¶
StartConfigured begins an import from the DataHub the configuration names, answering ErrNoSource when it names none.
type Skip ¶
type Skip struct {
URN string `json:"urn"`
Reason string `json:"reason"`
// Where names the platform table and column a rewrite skip was found in.
Where string `json:"where,omitempty"`
}
Skip is one thing the import or the rewrite did not carry over.
type Source ¶
type Source interface {
DatasetHeads(ctx context.Context) ([]DatasetHead, error)
Dataset(ctx context.Context, urn string) (*DatasetRecord, error)
Definitions(ctx context.Context) ([]Definition, error)
}
Source is what the import reads from DataHub.
type SourceConfig ¶
SourceConfig names the DataHub an import reads.
type SourceFactory ¶
SourceFactory opens what an import reads, for the DataHub an administrator named.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store keeps the import's runs and its mapping of DataHub URNs to catalog references in PostgreSQL.