Package org.odpi.openmetadata.adapters.connectors.integration.openlineage.postgres

Class PostgresOpenLineageLogStoreSchema

java.lang.Object
org.odpi.openmetadata.adapters.connectors.integration.openlineage.postgres.PostgresOpenLineageLogStoreSchema
public class PostgresOpenLineageLogStoreSchema extends Object
PostgresOpenLineageLogStoreSchema defines the tables of a PostgreSQL OpenLineage log store and the SQL that writes, purges and reads them. It is shared by the connector that writes the log store and the services (such as the Lovelace OpenLineage analysis services) that read it, so the table definitions live in one place.

The log store keeps each event twice:

  • ol_event - the event exactly as it was received, as JSONB, for analysis that is not anticipated here. The (run_id, event_time, event_type, event_hash) key means an event that is delivered twice (for example, when a Kafka topic is replayed) is stored once - and only the first delivery updates the digest.
  • a digest, maintained as each event arrives, that holds the shape the analysis services need:
    • ol_run - one row per run: its job, status, start and end time, and the egeria governance action facet.
    • ol_run_dataset - one row per run per input (I) or output (O) dataset with the statistics reported for it.
    • ol_assertion - each data quality assertion reported on an input dataset.
    • ol_test - each test reported in a run's test facet.

Reading a time window from the digest is a few indexed range scans that return one row per run, rather than a parse of every event, so the cost of an analysis follows the number of runs, not the number of events.

  • Field Details

  • Constructor Details

    • PostgresOpenLineageLogStoreSchema

      public PostgresOpenLineageLogStoreSchema(String schemaName)
      Create the definitions for a schema.
      Parameters:
      schemaName - name of the database schema
      Throws:
      IllegalArgumentException - the schema name is missing or can not be used as a quoted identifier

  • Method Details

    • isValidSchemaName

      public static boolean isValidSchemaName(String schemaName)
      Check that a schema name can be used. The name is always quoted in the SQL, so any characters are allowed other than a double quote (which would end the identifier) and the null character. PostgreSQL truncates identifiers longer than 63 bytes, which would quietly use a different schema, so those are rejected too.
      Parameters:
      schemaName - proposed name
      Returns:
      true if the name can be used

    • getSchemaName

      public String getSchemaName()
      Return the schema name.
      Returns:
      name

    • table

      public String table(String tableName)
      Return the qualified, quoted name of a table in this schema.
      Parameters:
      tableName - unqualified table name
      Returns:
      qualified name

    • getDDLStatements

      public List<String> getDDLStatements()
      Return the statements that create the schema, its tables and their indexes. Each one does nothing if the object already exists, so they are run each time the log store starts.
      Returns:
      list of DDL statements

    • getInsertEventSQL

      public String getInsertEventSQL()
      Insert a raw event. Returns no row if the event is already stored. Parameters: event_time, event_type, run_id, job_namespace, job_name, producer, event_hash, event.
      Returns:
      SQL

    • getUpsertRunSQL

      public String getUpsertRunSQL()
      Add an event to its run. The rules match those of the original file-based reader: the earliest START sets the start time; the latest COMPLETE, FAIL or ABORT sets the end time and the final status; RUNNING replaces only a START status; and the egeria governance action facet is kept from the last event that carried it. PostgreSQL's LEAST and GREATEST ignore nulls, so an event that does not set a time leaves it unchanged. Parameters: run_id, job_namespace, job_name, status, start_time, end_time, first_event_time, last_event_time, governance_action.
      Returns:
      SQL

    • getUpsertRunDatasetSQL

      public String getUpsertRunDatasetSQL()
      Record a dataset read or written by a run. The first sighting registers it (so a run that has only started is counted). A later sighting is used if it carries statistics or is not a START event: if it is the newest event for the dataset, by event time, it sets the record and any statistics it does not repeat are kept; if it is older (because events arrived out of order) it only fills in the values that are missing. So the result does not depend on the order in which the events arrive. Parameters: run_id, direction, dataset_namespace, dataset_name, event_time, event_type, row_count, size_bytes, file_count, lifecycle_state_change, subset_type, dataset_version.
      Returns:
      SQL

    • getInsertAssertionSQL

      public String getInsertAssertionSQL()
      Insert an assertion. Parameters: event_time, run_id, job_namespace, job_name, dataset_namespace, dataset_name, assertion, column_name, success, severity, assertion_name.
      Returns:
      SQL

    • getInsertTestSQL

      public String getInsertTestSQL()
      Insert a test. Parameters: event_time, run_id, job_namespace, job_name, test_name, status, test_type, severity.
      Returns:
      SQL

    • getPurgeSQL

      public List<String> getPurgeSQL()
      Return the statements that delete the records older than a cut-off time. Each takes the cut-off as its only parameter. A run is deleted once its last event is older than the cut-off, and its datasets go with it.
      Returns:
      list of SQL statements

    • getRunsInWindowSQL

      public String getRunsInWindowSQL()
      The runs that had an event in the window, oldest first.
      Returns:
      SQL

    • getRunDatasetsInWindowSQL

      public String getRunDatasetsInWindowSQL()
      The datasets read or written by the runs that had an event in the window, oldest first.
      Returns:
      SQL

    • getAssertionsInWindowSQL

      public String getAssertionsInWindowSQL()
      The assertions reported in the window, oldest first.
      Returns:
      SQL

    • getTestsInWindowSQL

      public String getTestsInWindowSQL()
      The tests reported in the window, oldest first.
      Returns:
      SQL

    • getEventSummaryInWindowSQL

      public String getEventSummaryInWindowSQL()
      The number of events in the window, and the times of the first and last.
      Returns:
      SQL

    • toStoredNamespace

      public static String toStoredNamespace(String namespace)
      Namespaces are optional in an event but part of the key in the digest tables, which can not hold null. A missing namespace is stored as an empty string.
      Parameters:
      namespace - namespace from the event
      Returns:
      value to store

    • fromStoredNamespace

      public static String fromStoredNamespace(String storedNamespace)
      Turn a stored namespace back into the value from the event.
      Parameters:
      storedNamespace - value from the database
      Returns:
      namespace, or null if the event had none