You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@polaris.apache.org by "iting0321 (via GitHub)" <gi...@apache.org> on 2026/09/11 23:14:33 UTC
[PR] OpenLineage: Add Persistence schema and JDBC models [polaris]
iting0321 opened a new pull request, #4827:
URL: https://github.com/apache/polaris/pull/4827
Please noted that this PR is **the last one** out of the three and still depends on #4705 and #4826
1. https://github.com/apache/polaris/pull/4705 (first dependent one)
2. https://github.com/apache/polaris/pull/4826 (second dependent one)
3. https://github.com/apache/polaris/pull/4827: **Current one**
<!--
- **Diff for early review:** https://github.com/iting0321/polaris/compare/ol-persistence-spi...iting0321:polaris:ol-jdbc-persistence-layer-2
-->
## Summary
Adds the JDBC persistence implementation for OpenLineage dataset and column lineage.
The JDBC backend now persists lineage datasets, dataset-level edges, and column-level edges, and supports querying depth-one upstream/downstream lineage graphs through the existing `LineagePersistence` contract.
## Changes
Add JDBC schema v5 for H2, PostgreSQL, and CockroachDB with:
- `lineage_datasets`
- `lineage_edges`
- `lineage_column_edges`
Implements JDBC-backed LineagePersistence, and persists dataset, edge, and column-level lineage by realm and OpenLineage identity. It also handles latest-event replacement semantics, realm cleanup, schema-version validation, and adds H2 plus PostgreSQL/CockroachDB test coverage.
## Testing
Added coverage in LineagePersistenceTest.java for dataset/edge/column upserts, timestamp monotonicity, stale upstream replacement, older-event protection, clearing upstream lineage, depth one lineage graph loading, and the pre-v5 schema guard.
Added coverage in LineagePersistenceJdbcIT.java for PostgreSQL and CockroachDB lineage CRUD smoke tests.
## Checklist
- [x] 🛡️ Don't disclose security issues! (contact security@apache.org)
- [x] 🔗 Clearly explained why the changes are needed, or linked related issues: Fixes #
- [x] 🧪 Added/updated tests with good coverage, or manually tested (and explained how)
- [x] 💡 Added comments for complex logic
- [ ] 🧾 Updated `CHANGELOG.md` (if needed)
- [ ] 📚 Updated documentation in `site/content/in-dev/unreleased` (if needed)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OpenLineage: Add Persistence schema and JDBC models [polaris]
Posted by "snazy (via GitHub)" <gi...@apache.org>.
snazy commented on PR #4827:
URL: https://github.com/apache/polaris/pull/4827#issuecomment-4767809073
I think this persistence piece, together with #4826, is too early to merge as-is.
The concern is not implementation quality, but commit point. Once we add the lineage persistence model and JDBC schema, changes to the lineage identity model, edge replacement semantics, column-lineage representation, or storage placement become schema/migration changes for users. That makes this much harder to revise than ordinary internal Java code.
The dev@ discussion still has open questions around at least:
- what exact query/use case the reduced local graph is supposed to support;
- whether “latest” edge replacement is the right semantic model;
- whether lineage persistence must be tied to the same JDBC/metastore schema;
- whether the REST/API and persistence pieces should be optional modules;
- what backend capability contract non-JDBC implementations are expected to satisfy.
I think we should settle those points before merging the persistence SPI/schema work. It is fine to keep iterating on the PR, but I would not treat it as merge-ready yet.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OpenLineage: Add Persistence schema and JDBC models [polaris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #4827: OpenLineage: Add Persistence schema and JDBC models
URL: https://github.com/apache/polaris/pull/4827
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OpenLineage: Add Persistence schema and JDBC models [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #4827:
URL: https://github.com/apache/polaris/pull/4827#discussion_r3488004794
##########
persistence/relational-jdbc/src/main/resources/postgres/schema-v5.sql:
##########
@@ -0,0 +1,344 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file--
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"). You may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Changes from v4:
+-- * Added lineage graph tables
+
+CREATE SCHEMA IF NOT EXISTS POLARIS_SCHEMA;
+SET search_path TO POLARIS_SCHEMA;
+
+CREATE TABLE IF NOT EXISTS version (
+ version_key TEXT PRIMARY KEY,
+ version_value INTEGER NOT NULL
+);
+INSERT INTO version (version_key, version_value)
+VALUES ('version', 5)
+ON CONFLICT (version_key) DO UPDATE
+SET version_value = EXCLUDED.version_value;
+COMMENT ON TABLE version IS 'the version of the JDBC schema in use';
+
+CREATE TABLE IF NOT EXISTS entities (
+ realm_id TEXT NOT NULL,
+ catalog_id BIGINT NOT NULL,
+ id BIGINT NOT NULL,
+ parent_id BIGINT NOT NULL,
+ name TEXT NOT NULL,
+ entity_version INT NOT NULL,
+ type_code INT NOT NULL,
+ sub_type_code INT NOT NULL,
+ create_timestamp BIGINT NOT NULL,
+ drop_timestamp BIGINT NOT NULL,
+ purge_timestamp BIGINT NOT NULL,
+ to_purge_timestamp BIGINT NOT NULL,
+ last_update_timestamp BIGINT NOT NULL,
+ properties JSONB not null default '{}'::JSONB,
+ internal_properties JSONB not null default '{}'::JSONB,
+ grant_records_version INT NOT NULL,
+ location_without_scheme TEXT,
+ PRIMARY KEY (realm_id, id),
+ CONSTRAINT constraint_name UNIQUE (realm_id, catalog_id, parent_id, type_code, name)
+);
+
+-- TODO: create indexes based on all query pattern.
+CREATE INDEX IF NOT EXISTS idx_entities ON entities (realm_id, catalog_id, id);
+CREATE INDEX IF NOT EXISTS idx_entities_catalog_id_id ON entities (catalog_id, id);
+CREATE INDEX IF NOT EXISTS idx_locations
+ ON entities USING btree (realm_id, parent_id, location_without_scheme)
+ WHERE location_without_scheme IS NOT NULL;
+
+COMMENT ON TABLE entities IS 'all the entities';
+
+COMMENT ON COLUMN entities.realm_id IS 'realm_id used for multi-tenancy';
+COMMENT ON COLUMN entities.catalog_id IS 'catalog id';
+COMMENT ON COLUMN entities.id IS 'entity id';
+COMMENT ON COLUMN entities.parent_id IS 'entity id of parent';
+COMMENT ON COLUMN entities.name IS 'entity name';
+COMMENT ON COLUMN entities.entity_version IS 'version of the entity';
+COMMENT ON COLUMN entities.type_code IS 'type code';
+COMMENT ON COLUMN entities.sub_type_code IS 'sub type of entity';
+COMMENT ON COLUMN entities.create_timestamp IS 'creation time of entity';
+COMMENT ON COLUMN entities.drop_timestamp IS 'time of drop of entity';
+COMMENT ON COLUMN entities.purge_timestamp IS 'time to start purging entity';
+COMMENT ON COLUMN entities.last_update_timestamp IS 'last time the entity is touched';
+COMMENT ON COLUMN entities.properties IS 'entities properties json';
+COMMENT ON COLUMN entities.internal_properties IS 'entities internal properties json';
+COMMENT ON COLUMN entities.grant_records_version IS 'the version of grant records change on the entity';
+
+CREATE TABLE IF NOT EXISTS grant_records (
+ realm_id TEXT NOT NULL,
+ securable_catalog_id BIGINT NOT NULL,
+ securable_id BIGINT NOT NULL,
+ grantee_catalog_id BIGINT NOT NULL,
+ grantee_id BIGINT NOT NULL,
+ privilege_code INTEGER,
+ PRIMARY KEY (realm_id, securable_catalog_id, securable_id, grantee_catalog_id, grantee_id, privilege_code)
+);
+
+COMMENT ON TABLE grant_records IS 'grant records for entities';
+
+COMMENT ON COLUMN grant_records.securable_catalog_id IS 'catalog id of the securable';
+COMMENT ON COLUMN grant_records.securable_id IS 'entity id of the securable';
+COMMENT ON COLUMN grant_records.grantee_catalog_id IS 'catalog id of the grantee';
+COMMENT ON COLUMN grant_records.grantee_id IS 'id of the grantee';
+COMMENT ON COLUMN grant_records.privilege_code IS 'privilege code';
+
+CREATE INDEX IF NOT EXISTS idx_grants_realm_grantee
+ ON grant_records (realm_id, grantee_id);
+CREATE INDEX IF NOT EXISTS idx_grants_realm_securable
+ ON grant_records (realm_id, securable_id);
+
+CREATE TABLE IF NOT EXISTS principal_authentication_data (
+ realm_id TEXT NOT NULL,
+ principal_id BIGINT NOT NULL,
+ principal_client_id VARCHAR(255) NOT NULL,
+ main_secret_hash VARCHAR(255) NOT NULL,
+ secondary_secret_hash VARCHAR(255) NOT NULL,
+ secret_salt VARCHAR(255) NOT NULL,
+ PRIMARY KEY (realm_id, principal_client_id)
+);
+
+COMMENT ON TABLE principal_authentication_data IS 'authentication data for client';
+
+CREATE TABLE IF NOT EXISTS policy_mapping_record (
+ realm_id TEXT NOT NULL,
+ target_catalog_id BIGINT NOT NULL,
+ target_id BIGINT NOT NULL,
+ policy_type_code INTEGER NOT NULL,
+ policy_catalog_id BIGINT NOT NULL,
+ policy_id BIGINT NOT NULL,
+ parameters JSONB NOT NULL DEFAULT '{}'::JSONB,
+ PRIMARY KEY (realm_id, target_catalog_id, target_id, policy_type_code, policy_catalog_id, policy_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_policy_mapping_record ON policy_mapping_record (realm_id, policy_type_code, policy_catalog_id, policy_id, target_catalog_id, target_id);
+
+CREATE TABLE IF NOT EXISTS events (
+ realm_id TEXT NOT NULL,
+ catalog_id TEXT NOT NULL,
+ event_id TEXT NOT NULL,
+ request_id TEXT,
+ event_type TEXT NOT NULL,
+ timestamp_ms BIGINT NOT NULL,
+ principal_name TEXT,
+ resource_type TEXT NOT NULL,
+ resource_identifier TEXT NOT NULL,
+ additional_properties JSONB NOT NULL DEFAULT '{}'::JSONB,
+ PRIMARY KEY (event_id)
+);
+
+-- Idempotency records (key-only idempotency; durable replay)
+CREATE TABLE IF NOT EXISTS idempotency_records (
+ realm_id TEXT NOT NULL,
+ idempotency_key TEXT NOT NULL,
+ operation_type TEXT NOT NULL,
+ resource_id TEXT NOT NULL, -- normalized request-derived resource identifier (not a generated entity id)
+
+ -- Finalization/replay
+ http_status INTEGER, -- NULL while IN_PROGRESS; set only on finalized 2xx/terminal 4xx
+ error_subtype TEXT, -- optional: e.g., already_exists, namespace_not_empty, idempotency_replay_failed
+ response_summary TEXT, -- minimal body to reproduce equivalent response (JSON string)
+ response_headers TEXT, -- small whitelisted headers to replay (JSON string)
+ finalized_at TIMESTAMP, -- when http_status was written
+
+ -- Liveness/ops
+ created_at TIMESTAMP NOT NULL,
+ updated_at TIMESTAMP NOT NULL,
+ heartbeat_at TIMESTAMP, -- updated by owner while IN_PROGRESS
+ executor_id TEXT, -- owner pod/worker id
+ expires_at TIMESTAMP,
+
+ PRIMARY KEY (realm_id, idempotency_key)
+);
+
+-- Helpful indexes
+CREATE INDEX IF NOT EXISTS idx_idemp_realm_expires
+ ON idempotency_records (realm_id, expires_at);
+
+-- ============================================================================
+-- SCAN METRICS REPORT TABLE
+-- ============================================================================
+
+CREATE TABLE IF NOT EXISTS scan_metrics_report (
Review Comment:
Thanks for your feedback! I’ll continue refactoring this as part of the "bigger change".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OpenLineage: Add Persistence schema and JDBC models [polaris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #4827:
URL: https://github.com/apache/polaris/pull/4827#issuecomment-5310938608
This PR is stale because it has been open 30 days with no activity. Remove stale label or comment or this will be closed in 5 days.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OpenLineage: Add Persistence schema and JDBC models [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #4827:
URL: https://github.com/apache/polaris/pull/4827#discussion_r3449654300
##########
persistence/relational-jdbc/src/main/resources/postgres/schema-v5.sql:
##########
@@ -0,0 +1,344 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file--
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"). You may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Changes from v4:
+-- * Added lineage graph tables
+
+CREATE SCHEMA IF NOT EXISTS POLARIS_SCHEMA;
+SET search_path TO POLARIS_SCHEMA;
+
+CREATE TABLE IF NOT EXISTS version (
+ version_key TEXT PRIMARY KEY,
+ version_value INTEGER NOT NULL
+);
+INSERT INTO version (version_key, version_value)
+VALUES ('version', 5)
+ON CONFLICT (version_key) DO UPDATE
+SET version_value = EXCLUDED.version_value;
+COMMENT ON TABLE version IS 'the version of the JDBC schema in use';
+
+CREATE TABLE IF NOT EXISTS entities (
+ realm_id TEXT NOT NULL,
+ catalog_id BIGINT NOT NULL,
+ id BIGINT NOT NULL,
+ parent_id BIGINT NOT NULL,
+ name TEXT NOT NULL,
+ entity_version INT NOT NULL,
+ type_code INT NOT NULL,
+ sub_type_code INT NOT NULL,
+ create_timestamp BIGINT NOT NULL,
+ drop_timestamp BIGINT NOT NULL,
+ purge_timestamp BIGINT NOT NULL,
+ to_purge_timestamp BIGINT NOT NULL,
+ last_update_timestamp BIGINT NOT NULL,
+ properties JSONB not null default '{}'::JSONB,
+ internal_properties JSONB not null default '{}'::JSONB,
+ grant_records_version INT NOT NULL,
+ location_without_scheme TEXT,
+ PRIMARY KEY (realm_id, id),
+ CONSTRAINT constraint_name UNIQUE (realm_id, catalog_id, parent_id, type_code, name)
+);
+
+-- TODO: create indexes based on all query pattern.
+CREATE INDEX IF NOT EXISTS idx_entities ON entities (realm_id, catalog_id, id);
+CREATE INDEX IF NOT EXISTS idx_entities_catalog_id_id ON entities (catalog_id, id);
+CREATE INDEX IF NOT EXISTS idx_locations
+ ON entities USING btree (realm_id, parent_id, location_without_scheme)
+ WHERE location_without_scheme IS NOT NULL;
+
+COMMENT ON TABLE entities IS 'all the entities';
+
+COMMENT ON COLUMN entities.realm_id IS 'realm_id used for multi-tenancy';
+COMMENT ON COLUMN entities.catalog_id IS 'catalog id';
+COMMENT ON COLUMN entities.id IS 'entity id';
+COMMENT ON COLUMN entities.parent_id IS 'entity id of parent';
+COMMENT ON COLUMN entities.name IS 'entity name';
+COMMENT ON COLUMN entities.entity_version IS 'version of the entity';
+COMMENT ON COLUMN entities.type_code IS 'type code';
+COMMENT ON COLUMN entities.sub_type_code IS 'sub type of entity';
+COMMENT ON COLUMN entities.create_timestamp IS 'creation time of entity';
+COMMENT ON COLUMN entities.drop_timestamp IS 'time of drop of entity';
+COMMENT ON COLUMN entities.purge_timestamp IS 'time to start purging entity';
+COMMENT ON COLUMN entities.last_update_timestamp IS 'last time the entity is touched';
+COMMENT ON COLUMN entities.properties IS 'entities properties json';
+COMMENT ON COLUMN entities.internal_properties IS 'entities internal properties json';
+COMMENT ON COLUMN entities.grant_records_version IS 'the version of grant records change on the entity';
+
+CREATE TABLE IF NOT EXISTS grant_records (
+ realm_id TEXT NOT NULL,
+ securable_catalog_id BIGINT NOT NULL,
+ securable_id BIGINT NOT NULL,
+ grantee_catalog_id BIGINT NOT NULL,
+ grantee_id BIGINT NOT NULL,
+ privilege_code INTEGER,
+ PRIMARY KEY (realm_id, securable_catalog_id, securable_id, grantee_catalog_id, grantee_id, privilege_code)
+);
+
+COMMENT ON TABLE grant_records IS 'grant records for entities';
+
+COMMENT ON COLUMN grant_records.securable_catalog_id IS 'catalog id of the securable';
+COMMENT ON COLUMN grant_records.securable_id IS 'entity id of the securable';
+COMMENT ON COLUMN grant_records.grantee_catalog_id IS 'catalog id of the grantee';
+COMMENT ON COLUMN grant_records.grantee_id IS 'id of the grantee';
+COMMENT ON COLUMN grant_records.privilege_code IS 'privilege code';
+
+CREATE INDEX IF NOT EXISTS idx_grants_realm_grantee
+ ON grant_records (realm_id, grantee_id);
+CREATE INDEX IF NOT EXISTS idx_grants_realm_securable
+ ON grant_records (realm_id, securable_id);
+
+CREATE TABLE IF NOT EXISTS principal_authentication_data (
+ realm_id TEXT NOT NULL,
+ principal_id BIGINT NOT NULL,
+ principal_client_id VARCHAR(255) NOT NULL,
+ main_secret_hash VARCHAR(255) NOT NULL,
+ secondary_secret_hash VARCHAR(255) NOT NULL,
+ secret_salt VARCHAR(255) NOT NULL,
+ PRIMARY KEY (realm_id, principal_client_id)
+);
+
+COMMENT ON TABLE principal_authentication_data IS 'authentication data for client';
+
+CREATE TABLE IF NOT EXISTS policy_mapping_record (
+ realm_id TEXT NOT NULL,
+ target_catalog_id BIGINT NOT NULL,
+ target_id BIGINT NOT NULL,
+ policy_type_code INTEGER NOT NULL,
+ policy_catalog_id BIGINT NOT NULL,
+ policy_id BIGINT NOT NULL,
+ parameters JSONB NOT NULL DEFAULT '{}'::JSONB,
+ PRIMARY KEY (realm_id, target_catalog_id, target_id, policy_type_code, policy_catalog_id, policy_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_policy_mapping_record ON policy_mapping_record (realm_id, policy_type_code, policy_catalog_id, policy_id, target_catalog_id, target_id);
+
+CREATE TABLE IF NOT EXISTS events (
+ realm_id TEXT NOT NULL,
+ catalog_id TEXT NOT NULL,
+ event_id TEXT NOT NULL,
+ request_id TEXT,
+ event_type TEXT NOT NULL,
+ timestamp_ms BIGINT NOT NULL,
+ principal_name TEXT,
+ resource_type TEXT NOT NULL,
+ resource_identifier TEXT NOT NULL,
+ additional_properties JSONB NOT NULL DEFAULT '{}'::JSONB,
+ PRIMARY KEY (event_id)
+);
+
+-- Idempotency records (key-only idempotency; durable replay)
+CREATE TABLE IF NOT EXISTS idempotency_records (
+ realm_id TEXT NOT NULL,
+ idempotency_key TEXT NOT NULL,
+ operation_type TEXT NOT NULL,
+ resource_id TEXT NOT NULL, -- normalized request-derived resource identifier (not a generated entity id)
+
+ -- Finalization/replay
+ http_status INTEGER, -- NULL while IN_PROGRESS; set only on finalized 2xx/terminal 4xx
+ error_subtype TEXT, -- optional: e.g., already_exists, namespace_not_empty, idempotency_replay_failed
+ response_summary TEXT, -- minimal body to reproduce equivalent response (JSON string)
+ response_headers TEXT, -- small whitelisted headers to replay (JSON string)
+ finalized_at TIMESTAMP, -- when http_status was written
+
+ -- Liveness/ops
+ created_at TIMESTAMP NOT NULL,
+ updated_at TIMESTAMP NOT NULL,
+ heartbeat_at TIMESTAMP, -- updated by owner while IN_PROGRESS
+ executor_id TEXT, -- owner pod/worker id
+ expires_at TIMESTAMP,
+
+ PRIMARY KEY (realm_id, idempotency_key)
+);
+
+-- Helpful indexes
+CREATE INDEX IF NOT EXISTS idx_idemp_realm_expires
+ ON idempotency_records (realm_id, expires_at);
+
+-- ============================================================================
+-- SCAN METRICS REPORT TABLE
+-- ============================================================================
+
+CREATE TABLE IF NOT EXISTS scan_metrics_report (
Review Comment:
Hi @dimas-b, I think there are two possible levels of change here:
**Small change:** keep the existing JDBC bootstrap contract: Polaris still asks `DatabaseType` for one top-level schema resource, e.g. `postgres/schema-v5.sql`, and bootstrap still executes a single SQL stream.
The change would be inside the schema resource loader: before returning that stream, it would expand fragment directives in `schema-v5.sql`. So `schema-v5.sql` can keep the existing core/events/metrics DDL inline, and replace only the OpenLineage section with:
```sql
-- POLARIS_SCHEMA_FRAGMENT: schema/lineage/lineage-v1.sql
```
The loader would resolve that to postgres/schema/lineage/lineage-v1.sql, concatenate it into the SQL stream, and then bootstrap executes the combined result.
The architecture will look like:
```
postgres/
├── schema-v5.sql
└── schema/
└── lineage/
└── lineage-v1.sql
```
**Bigger change:** make bootstrap understand separately versioned schema components, e.g. `core = v5`, `lineage = v1`. That would let lineage evolve independently from the MetaStore schema, but it requires changing how we record schema versions, how bootstrap decides what to run, and eventually how upgrades/migrations work per component.
For this PR, I’d prefer the small change: split the DDL by feature while preserving the current single-version bootstrap behavior. Then we can leave independently versioned persistence components as a follow-up design.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OpenLineage: Add Persistence schema and JDBC models [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on PR #4827:
URL: https://github.com/apache/polaris/pull/4827#issuecomment-4768677997
> I think this persistence piece, together with #4826, is too early to merge as-is.
>
> The concern is not implementation quality, but commit point. Once we add the lineage persistence model and JDBC schema, changes to the lineage identity model, edge replacement semantics, column-lineage representation, or storage placement become schema/migration changes for users. That makes this much harder to revise than ordinary internal Java code.
>
> The dev@ discussion still has open questions around at least:
>
> * what exact query/use case the reduced local graph is supposed to support;
> * whether “latest” edge replacement is the right semantic model;
> * whether lineage persistence must be tied to the same JDBC/metastore schema;
> * whether the REST/API and persistence pieces should be optional modules;
> * what backend capability contract non-JDBC implementations are expected to satisfy.
>
> I think we should settle those points before merging the persistence SPI/schema work. It is fine to keep iterating on the PR, but I would not treat it as merge-ready yet.
Hi @snazy , thanks for clarifying. I understand your concern about committing to the persistence model and schema before the remaining design questions are resolved. I’ll continue following the dev@ discussion and update the PR based on the resulting consensus.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OpenLineage: Add Persistence schema and JDBC models [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4827:
URL: https://github.com/apache/polaris/pull/4827#discussion_r3477947435
##########
persistence/relational-jdbc/src/main/resources/postgres/schema-v5.sql:
##########
@@ -0,0 +1,344 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file--
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"). You may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Changes from v4:
+-- * Added lineage graph tables
+
+CREATE SCHEMA IF NOT EXISTS POLARIS_SCHEMA;
+SET search_path TO POLARIS_SCHEMA;
+
+CREATE TABLE IF NOT EXISTS version (
+ version_key TEXT PRIMARY KEY,
+ version_value INTEGER NOT NULL
+);
+INSERT INTO version (version_key, version_value)
+VALUES ('version', 5)
+ON CONFLICT (version_key) DO UPDATE
+SET version_value = EXCLUDED.version_value;
+COMMENT ON TABLE version IS 'the version of the JDBC schema in use';
+
+CREATE TABLE IF NOT EXISTS entities (
+ realm_id TEXT NOT NULL,
+ catalog_id BIGINT NOT NULL,
+ id BIGINT NOT NULL,
+ parent_id BIGINT NOT NULL,
+ name TEXT NOT NULL,
+ entity_version INT NOT NULL,
+ type_code INT NOT NULL,
+ sub_type_code INT NOT NULL,
+ create_timestamp BIGINT NOT NULL,
+ drop_timestamp BIGINT NOT NULL,
+ purge_timestamp BIGINT NOT NULL,
+ to_purge_timestamp BIGINT NOT NULL,
+ last_update_timestamp BIGINT NOT NULL,
+ properties JSONB not null default '{}'::JSONB,
+ internal_properties JSONB not null default '{}'::JSONB,
+ grant_records_version INT NOT NULL,
+ location_without_scheme TEXT,
+ PRIMARY KEY (realm_id, id),
+ CONSTRAINT constraint_name UNIQUE (realm_id, catalog_id, parent_id, type_code, name)
+);
+
+-- TODO: create indexes based on all query pattern.
+CREATE INDEX IF NOT EXISTS idx_entities ON entities (realm_id, catalog_id, id);
+CREATE INDEX IF NOT EXISTS idx_entities_catalog_id_id ON entities (catalog_id, id);
+CREATE INDEX IF NOT EXISTS idx_locations
+ ON entities USING btree (realm_id, parent_id, location_without_scheme)
+ WHERE location_without_scheme IS NOT NULL;
+
+COMMENT ON TABLE entities IS 'all the entities';
+
+COMMENT ON COLUMN entities.realm_id IS 'realm_id used for multi-tenancy';
+COMMENT ON COLUMN entities.catalog_id IS 'catalog id';
+COMMENT ON COLUMN entities.id IS 'entity id';
+COMMENT ON COLUMN entities.parent_id IS 'entity id of parent';
+COMMENT ON COLUMN entities.name IS 'entity name';
+COMMENT ON COLUMN entities.entity_version IS 'version of the entity';
+COMMENT ON COLUMN entities.type_code IS 'type code';
+COMMENT ON COLUMN entities.sub_type_code IS 'sub type of entity';
+COMMENT ON COLUMN entities.create_timestamp IS 'creation time of entity';
+COMMENT ON COLUMN entities.drop_timestamp IS 'time of drop of entity';
+COMMENT ON COLUMN entities.purge_timestamp IS 'time to start purging entity';
+COMMENT ON COLUMN entities.last_update_timestamp IS 'last time the entity is touched';
+COMMENT ON COLUMN entities.properties IS 'entities properties json';
+COMMENT ON COLUMN entities.internal_properties IS 'entities internal properties json';
+COMMENT ON COLUMN entities.grant_records_version IS 'the version of grant records change on the entity';
+
+CREATE TABLE IF NOT EXISTS grant_records (
+ realm_id TEXT NOT NULL,
+ securable_catalog_id BIGINT NOT NULL,
+ securable_id BIGINT NOT NULL,
+ grantee_catalog_id BIGINT NOT NULL,
+ grantee_id BIGINT NOT NULL,
+ privilege_code INTEGER,
+ PRIMARY KEY (realm_id, securable_catalog_id, securable_id, grantee_catalog_id, grantee_id, privilege_code)
+);
+
+COMMENT ON TABLE grant_records IS 'grant records for entities';
+
+COMMENT ON COLUMN grant_records.securable_catalog_id IS 'catalog id of the securable';
+COMMENT ON COLUMN grant_records.securable_id IS 'entity id of the securable';
+COMMENT ON COLUMN grant_records.grantee_catalog_id IS 'catalog id of the grantee';
+COMMENT ON COLUMN grant_records.grantee_id IS 'id of the grantee';
+COMMENT ON COLUMN grant_records.privilege_code IS 'privilege code';
+
+CREATE INDEX IF NOT EXISTS idx_grants_realm_grantee
+ ON grant_records (realm_id, grantee_id);
+CREATE INDEX IF NOT EXISTS idx_grants_realm_securable
+ ON grant_records (realm_id, securable_id);
+
+CREATE TABLE IF NOT EXISTS principal_authentication_data (
+ realm_id TEXT NOT NULL,
+ principal_id BIGINT NOT NULL,
+ principal_client_id VARCHAR(255) NOT NULL,
+ main_secret_hash VARCHAR(255) NOT NULL,
+ secondary_secret_hash VARCHAR(255) NOT NULL,
+ secret_salt VARCHAR(255) NOT NULL,
+ PRIMARY KEY (realm_id, principal_client_id)
+);
+
+COMMENT ON TABLE principal_authentication_data IS 'authentication data for client';
+
+CREATE TABLE IF NOT EXISTS policy_mapping_record (
+ realm_id TEXT NOT NULL,
+ target_catalog_id BIGINT NOT NULL,
+ target_id BIGINT NOT NULL,
+ policy_type_code INTEGER NOT NULL,
+ policy_catalog_id BIGINT NOT NULL,
+ policy_id BIGINT NOT NULL,
+ parameters JSONB NOT NULL DEFAULT '{}'::JSONB,
+ PRIMARY KEY (realm_id, target_catalog_id, target_id, policy_type_code, policy_catalog_id, policy_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_policy_mapping_record ON policy_mapping_record (realm_id, policy_type_code, policy_catalog_id, policy_id, target_catalog_id, target_id);
+
+CREATE TABLE IF NOT EXISTS events (
+ realm_id TEXT NOT NULL,
+ catalog_id TEXT NOT NULL,
+ event_id TEXT NOT NULL,
+ request_id TEXT,
+ event_type TEXT NOT NULL,
+ timestamp_ms BIGINT NOT NULL,
+ principal_name TEXT,
+ resource_type TEXT NOT NULL,
+ resource_identifier TEXT NOT NULL,
+ additional_properties JSONB NOT NULL DEFAULT '{}'::JSONB,
+ PRIMARY KEY (event_id)
+);
+
+-- Idempotency records (key-only idempotency; durable replay)
+CREATE TABLE IF NOT EXISTS idempotency_records (
+ realm_id TEXT NOT NULL,
+ idempotency_key TEXT NOT NULL,
+ operation_type TEXT NOT NULL,
+ resource_id TEXT NOT NULL, -- normalized request-derived resource identifier (not a generated entity id)
+
+ -- Finalization/replay
+ http_status INTEGER, -- NULL while IN_PROGRESS; set only on finalized 2xx/terminal 4xx
+ error_subtype TEXT, -- optional: e.g., already_exists, namespace_not_empty, idempotency_replay_failed
+ response_summary TEXT, -- minimal body to reproduce equivalent response (JSON string)
+ response_headers TEXT, -- small whitelisted headers to replay (JSON string)
+ finalized_at TIMESTAMP, -- when http_status was written
+
+ -- Liveness/ops
+ created_at TIMESTAMP NOT NULL,
+ updated_at TIMESTAMP NOT NULL,
+ heartbeat_at TIMESTAMP, -- updated by owner while IN_PROGRESS
+ executor_id TEXT, -- owner pod/worker id
+ expires_at TIMESTAMP,
+
+ PRIMARY KEY (realm_id, idempotency_key)
+);
+
+-- Helpful indexes
+CREATE INDEX IF NOT EXISTS idx_idemp_realm_expires
+ ON idempotency_records (realm_id, expires_at);
+
+-- ============================================================================
+-- SCAN METRICS REPORT TABLE
+-- ============================================================================
+
+CREATE TABLE IF NOT EXISTS scan_metrics_report (
Review Comment:
My preference is for the "bigger change". I understand, it's more work, but at the same time it allows greater flexibility in new (OL) schema evolution.
Also, users not interested in the OL feature will not have to deal with OL tables.
Given the recent influx of persistence-related features (OL, OSI, Metrics, Events). I think it is time to revisit and enhance schema management and bootstrap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OpenLineage: Add Persistence schema and JDBC models [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #4827:
URL: https://github.com/apache/polaris/pull/4827#discussion_r3487974396
##########
persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java:
##########
@@ -1354,4 +1392,500 @@ private void writeCommitMetricsReport(@NonNull ModelCommitMetricsReport report)
String.format("Failed to write commit metrics report due to %s", e.getMessage()), e);
}
}
+
+ // ============================================================================
+ // LineagePersistence Implementation
+ // ============================================================================
+
+ @Override
+ public void upsertDatasets(RealmContext realmContext, List<LineageDataset> datasets) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long nowMillis = Instant.now().toEpochMilli();
+ for (LineageDataset dataset : datasets) {
+ long datasetId =
+ lookupLineageDataset(realmId, dataset)
+ .map(ModelLineageDataset::getDatasetId)
+ .orElseGet(IdGenerator.getIdGenerator()::nextId);
+ ModelLineageDataset model =
+ ModelLineageDataset.fromDataset(dataset, realmId, datasetId, nowMillis);
+ try {
+ datasourceOperations.executeUpdate(generateLineageDatasetUpsert(model));
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to upsert lineage dataset due to %s", e.getMessage()), e);
+ }
+ }
+ }
+
+ @Override
+ public void replaceDatasetEdges(
+ RealmContext realmContext,
+ List<LineageDataset> targetDatasets,
+ List<LineageEdge> edges,
+ Instant lastEventAt) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long lastEventAtMillis = lastEventAt.toEpochMilli();
+ Map<Long, List<ModelLineageEdge>> edgesByTargetDatasetId = new LinkedHashMap<>();
+ for (LineageDataset targetDataset : targetDatasets) {
+ edgesByTargetDatasetId.putIfAbsent(
+ requireLineageDatasetId(realmId, targetDataset), new ArrayList<>());
+ }
+ for (LineageEdge edge : edges) {
+ ModelLineageEdge model =
+ ModelLineageEdge.fromIds(
+ realmId,
+ requireLineageDatasetId(realmId, edge.source()),
+ requireLineageDatasetId(realmId, edge.target()),
+ lastEventAtMillis);
+ edgesByTargetDatasetId
+ .computeIfAbsent(model.getTargetDatasetId(), ignored -> new ArrayList<>())
+ .add(model);
+ }
+
+ try {
+ datasourceOperations.runWithinTransaction(
+ connection -> {
+ for (Map.Entry<Long, List<ModelLineageEdge>> entry :
+ edgesByTargetDatasetId.entrySet()) {
+ long targetDatasetId = entry.getKey();
+ Optional<Long> currentLastEventAt =
+ loadLineageDatasetLastEventAt(connection, realmId, targetDatasetId);
+ if (currentLastEventAt.isPresent() && currentLastEventAt.get() > lastEventAtMillis) {
+ continue;
+ }
+
+ List<Long> sourceDatasetIds =
+ entry.getValue().stream()
+ .map(ModelLineageEdge::getSourceDatasetId)
+ .distinct()
+ .toList();
+ datasourceOperations.execute(
+ connection,
+ generateDeleteStaleLineageEdges(realmId, targetDatasetId, sourceDatasetIds));
+ datasourceOperations.execute(
+ connection,
+ generateDeleteStaleLineageColumnEdges(
+ realmId, targetDatasetId, sourceDatasetIds));
+ for (ModelLineageEdge model : entry.getValue()) {
+ datasourceOperations.execute(connection, generateLineageEdgeUpsert(model));
+ }
+ datasourceOperations.execute(
+ connection,
+ generateLineageDatasetLastEventAtUpdate(
+ realmId, targetDatasetId, lastEventAtMillis));
+ }
+ return true;
+ });
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to upsert lineage edge due to %s", e.getMessage()), e);
+ }
+ }
+
+ private Optional<Long> loadLineageDatasetLastEventAt(
+ Connection connection, String realmId, long targetDatasetId) throws SQLException {
+ String table = QueryGenerator.getFullyQualifiedTableName(ModelLineageDataset.TABLE_NAME);
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "SELECT last_lineage_event_at FROM "
+ + table
+ + " WHERE realm_id = ? AND dataset_id = ?")) {
+ statement.setString(1, realmId);
+ statement.setLong(2, targetDatasetId);
+ try (ResultSet resultSet = statement.executeQuery()) {
+ if (!resultSet.next()) {
+ return Optional.empty();
+ }
+ long value = resultSet.getLong(1);
+ return resultSet.wasNull() ? Optional.empty() : Optional.of(value);
+ }
+ }
+ }
+
+ @Override
+ public void upsertColumnEdges(
+ RealmContext realmContext, List<LineageColumnEdge> columnEdges, Instant lastEventAt) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long lastEventAtMillis = lastEventAt.toEpochMilli();
+ for (LineageColumnEdge columnEdge : columnEdges) {
+ ModelLineageDataset targetDataset =
+ requireLineageDataset(realmId, columnEdge.target().dataset());
+ Long currentLastEventAt = targetDataset.getLastLineageEventAt();
+ if (currentLastEventAt != null && currentLastEventAt > lastEventAtMillis) {
+ continue;
+ }
+
+ ModelLineageColumnEdge model =
+ ModelLineageColumnEdge.fromIds(
+ realmId,
+ requireLineageDatasetId(realmId, columnEdge.source().dataset()),
+ columnEdge.source().field(),
+ targetDataset.getDatasetId(),
+ columnEdge.target().field(),
+ lastEventAtMillis);
+ try {
+ datasourceOperations.executeUpdate(generateLineageColumnEdgeUpsert(model));
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to upsert lineage column edge due to %s", e.getMessage()), e);
+ }
+ }
+ }
+
+ @Override
+ public LineageGraph loadLineage(RealmContext realmContext, LineageQueryRequest request) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ Optional<ModelLineageDataset> requested =
+ lookupLineageDatasetByNodeId(realmId, request.nodeId());
+ if (requested.isEmpty()) {
+ return new LineageGraph(
+ new LineageNode(request.nodeId(), LineageNodeType.DATASET, null, true),
+ List.of(),
+ List.of());
+ }
+
+ ModelLineageDataset dataset = requested.get();
+ boolean includeColumns = request.granularity() == LineageGranularity.COLUMN;
+ List<LineageNode> upstream =
+ request.direction() == LineageDirection.UPSTREAM
+ || request.direction() == LineageDirection.BOTH
+ ? loadAdjacentLineageNodes(realmId, dataset.getDatasetId(), true, includeColumns)
+ : List.of();
+ List<LineageNode> downstream =
+ request.direction() == LineageDirection.DOWNSTREAM
+ || request.direction() == LineageDirection.BOTH
+ ? loadAdjacentLineageNodes(realmId, dataset.getDatasetId(), false, includeColumns)
+ : List.of();
+
+ return new LineageGraph(toLineageNode(dataset, List.of()), upstream, downstream);
+ }
+
+ private Optional<ModelLineageDataset> lookupLineageDataset(
+ String realmId, LineageDataset dataset) {
+ String table = QueryGenerator.getFullyQualifiedTableName(ModelLineageDataset.TABLE_NAME);
+ PreparedQuery query =
+ new PreparedQuery(
+ "SELECT realm_id, dataset_id, catalog, namespace, name, polaris_entity_id, last_lineage_event_at, created_at, updated_at "
+ + "FROM "
+ + table
+ + " WHERE realm_id = ? AND namespace = ? AND name = ?",
+ List.of(realmId, dataset.namespace(), dataset.name()));
Review Comment:
fixed
##########
persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java:
##########
@@ -1354,4 +1392,500 @@ private void writeCommitMetricsReport(@NonNull ModelCommitMetricsReport report)
String.format("Failed to write commit metrics report due to %s", e.getMessage()), e);
}
}
+
+ // ============================================================================
+ // LineagePersistence Implementation
+ // ============================================================================
+
+ @Override
+ public void upsertDatasets(RealmContext realmContext, List<LineageDataset> datasets) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long nowMillis = Instant.now().toEpochMilli();
+ for (LineageDataset dataset : datasets) {
+ long datasetId =
+ lookupLineageDataset(realmId, dataset)
+ .map(ModelLineageDataset::getDatasetId)
+ .orElseGet(IdGenerator.getIdGenerator()::nextId);
+ ModelLineageDataset model =
+ ModelLineageDataset.fromDataset(dataset, realmId, datasetId, nowMillis);
+ try {
+ datasourceOperations.executeUpdate(generateLineageDatasetUpsert(model));
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to upsert lineage dataset due to %s", e.getMessage()), e);
+ }
+ }
+ }
+
+ @Override
+ public void replaceDatasetEdges(
+ RealmContext realmContext,
+ List<LineageDataset> targetDatasets,
+ List<LineageEdge> edges,
+ Instant lastEventAt) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long lastEventAtMillis = lastEventAt.toEpochMilli();
+ Map<Long, List<ModelLineageEdge>> edgesByTargetDatasetId = new LinkedHashMap<>();
+ for (LineageDataset targetDataset : targetDatasets) {
+ edgesByTargetDatasetId.putIfAbsent(
+ requireLineageDatasetId(realmId, targetDataset), new ArrayList<>());
+ }
+ for (LineageEdge edge : edges) {
+ ModelLineageEdge model =
+ ModelLineageEdge.fromIds(
+ realmId,
+ requireLineageDatasetId(realmId, edge.source()),
+ requireLineageDatasetId(realmId, edge.target()),
+ lastEventAtMillis);
+ edgesByTargetDatasetId
+ .computeIfAbsent(model.getTargetDatasetId(), ignored -> new ArrayList<>())
+ .add(model);
+ }
+
+ try {
+ datasourceOperations.runWithinTransaction(
+ connection -> {
+ for (Map.Entry<Long, List<ModelLineageEdge>> entry :
+ edgesByTargetDatasetId.entrySet()) {
+ long targetDatasetId = entry.getKey();
+ Optional<Long> currentLastEventAt =
+ loadLineageDatasetLastEventAt(connection, realmId, targetDatasetId);
+ if (currentLastEventAt.isPresent() && currentLastEventAt.get() > lastEventAtMillis) {
+ continue;
+ }
+
+ List<Long> sourceDatasetIds =
+ entry.getValue().stream()
+ .map(ModelLineageEdge::getSourceDatasetId)
+ .distinct()
+ .toList();
+ datasourceOperations.execute(
+ connection,
+ generateDeleteStaleLineageEdges(realmId, targetDatasetId, sourceDatasetIds));
+ datasourceOperations.execute(
+ connection,
+ generateDeleteStaleLineageColumnEdges(
+ realmId, targetDatasetId, sourceDatasetIds));
+ for (ModelLineageEdge model : entry.getValue()) {
+ datasourceOperations.execute(connection, generateLineageEdgeUpsert(model));
+ }
+ datasourceOperations.execute(
+ connection,
+ generateLineageDatasetLastEventAtUpdate(
+ realmId, targetDatasetId, lastEventAtMillis));
+ }
+ return true;
+ });
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to upsert lineage edge due to %s", e.getMessage()), e);
+ }
+ }
+
+ private Optional<Long> loadLineageDatasetLastEventAt(
+ Connection connection, String realmId, long targetDatasetId) throws SQLException {
+ String table = QueryGenerator.getFullyQualifiedTableName(ModelLineageDataset.TABLE_NAME);
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "SELECT last_lineage_event_at FROM "
+ + table
+ + " WHERE realm_id = ? AND dataset_id = ?")) {
+ statement.setString(1, realmId);
+ statement.setLong(2, targetDatasetId);
+ try (ResultSet resultSet = statement.executeQuery()) {
+ if (!resultSet.next()) {
+ return Optional.empty();
+ }
+ long value = resultSet.getLong(1);
+ return resultSet.wasNull() ? Optional.empty() : Optional.of(value);
+ }
+ }
+ }
+
+ @Override
+ public void upsertColumnEdges(
+ RealmContext realmContext, List<LineageColumnEdge> columnEdges, Instant lastEventAt) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long lastEventAtMillis = lastEventAt.toEpochMilli();
+ for (LineageColumnEdge columnEdge : columnEdges) {
+ ModelLineageDataset targetDataset =
+ requireLineageDataset(realmId, columnEdge.target().dataset());
+ Long currentLastEventAt = targetDataset.getLastLineageEventAt();
+ if (currentLastEventAt != null && currentLastEventAt > lastEventAtMillis) {
+ continue;
+ }
+
+ ModelLineageColumnEdge model =
+ ModelLineageColumnEdge.fromIds(
+ realmId,
+ requireLineageDatasetId(realmId, columnEdge.source().dataset()),
+ columnEdge.source().field(),
+ targetDataset.getDatasetId(),
+ columnEdge.target().field(),
+ lastEventAtMillis);
+ try {
+ datasourceOperations.executeUpdate(generateLineageColumnEdgeUpsert(model));
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to upsert lineage column edge due to %s", e.getMessage()), e);
+ }
+ }
+ }
+
+ @Override
+ public LineageGraph loadLineage(RealmContext realmContext, LineageQueryRequest request) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ Optional<ModelLineageDataset> requested =
+ lookupLineageDatasetByNodeId(realmId, request.nodeId());
+ if (requested.isEmpty()) {
+ return new LineageGraph(
+ new LineageNode(request.nodeId(), LineageNodeType.DATASET, null, true),
+ List.of(),
+ List.of());
+ }
+
+ ModelLineageDataset dataset = requested.get();
+ boolean includeColumns = request.granularity() == LineageGranularity.COLUMN;
+ List<LineageNode> upstream =
+ request.direction() == LineageDirection.UPSTREAM
+ || request.direction() == LineageDirection.BOTH
+ ? loadAdjacentLineageNodes(realmId, dataset.getDatasetId(), true, includeColumns)
+ : List.of();
+ List<LineageNode> downstream =
+ request.direction() == LineageDirection.DOWNSTREAM
+ || request.direction() == LineageDirection.BOTH
+ ? loadAdjacentLineageNodes(realmId, dataset.getDatasetId(), false, includeColumns)
+ : List.of();
+
+ return new LineageGraph(toLineageNode(dataset, List.of()), upstream, downstream);
+ }
+
+ private Optional<ModelLineageDataset> lookupLineageDataset(
+ String realmId, LineageDataset dataset) {
+ String table = QueryGenerator.getFullyQualifiedTableName(ModelLineageDataset.TABLE_NAME);
+ PreparedQuery query =
+ new PreparedQuery(
+ "SELECT realm_id, dataset_id, catalog, namespace, name, polaris_entity_id, last_lineage_event_at, created_at, updated_at "
+ + "FROM "
+ + table
+ + " WHERE realm_id = ? AND namespace = ? AND name = ?",
+ List.of(realmId, dataset.namespace(), dataset.name()));
+ try {
+ List<ModelLineageDataset> results =
+ datasourceOperations.executeSelect(query, ModelLineageDataset.CONVERTER);
+ return results.stream().findFirst();
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to load lineage dataset due to %s", e.getMessage()), e);
+ }
+ }
+
+ private Optional<ModelLineageDataset> lookupLineageDatasetByNodeId(
+ String realmId, String nodeId) {
+ return parseLineageDatasetNodeId(nodeId)
+ .flatMap(dataset -> lookupLineageDataset(realmId, dataset));
+ }
+
+ private Optional<LineageDataset> parseLineageDatasetNodeId(String nodeId) {
+ String prefix = "dataset:";
+ if (!nodeId.startsWith(prefix)) {
+ return Optional.empty();
+ }
+ String identity = nodeId.substring(prefix.length());
+ int catalogSeparator = identity.lastIndexOf(':');
+ int nameSeparator = identity.lastIndexOf('.');
+ if (catalogSeparator < 0 || nameSeparator <= catalogSeparator + 1) {
+ return Optional.empty();
+ }
+ String catalog = identity.substring(0, catalogSeparator);
+ String namespace = identity.substring(catalogSeparator + 1, nameSeparator);
+ String name = identity.substring(nameSeparator + 1);
+ return Optional.of(new LineageDataset(catalog, namespace, name));
+ }
+
+ private long requireLineageDatasetId(String realmId, LineageDataset dataset) {
+ return requireLineageDataset(realmId, dataset).getDatasetId();
+ }
+
+ private ModelLineageDataset requireLineageDataset(String realmId, LineageDataset dataset) {
+ return lookupLineageDataset(realmId, dataset)
+ .orElseThrow(
+ () ->
+ new IllegalArgumentException(
+ String.format(
+ "Lineage dataset '%s.%s' does not exist in realm '%s'. Call upsertDatasets before writing edges.",
+ dataset.namespace(), dataset.name(), realmId)));
+ }
+
+ private List<LineageNode> loadAdjacentLineageNodes(
+ String realmId, long datasetId, boolean upstream, boolean includeColumns) {
+ String edgesTable = QueryGenerator.getFullyQualifiedTableName(ModelLineageEdge.TABLE_NAME);
+ String datasetsTable =
+ QueryGenerator.getFullyQualifiedTableName(ModelLineageDataset.TABLE_NAME);
+ String adjacentEdgeColumn = upstream ? "source_dataset_id" : "target_dataset_id";
+ String requestedEdgeColumn = upstream ? "target_dataset_id" : "source_dataset_id";
+ PreparedQuery query =
+ new PreparedQuery(
+ "SELECT d.realm_id, d.dataset_id, d.catalog, d.namespace, d.name, d.polaris_entity_id, d.last_lineage_event_at, d.created_at, d.updated_at "
+ + "FROM "
+ + edgesTable
+ + " e JOIN "
+ + datasetsTable
+ + " d ON d.realm_id = e.realm_id AND d.dataset_id = e."
+ + adjacentEdgeColumn
+ + " WHERE e.realm_id = ? AND e."
+ + requestedEdgeColumn
+ + " = ?",
+ List.of(realmId, datasetId));
+ try {
+ List<ModelLineageDataset> datasets =
+ datasourceOperations.executeSelect(query, ModelLineageDataset.CONVERTER);
+ return datasets.stream()
+ .map(
+ dataset ->
+ toLineageNode(
+ dataset,
+ includeColumns
+ ? loadFieldMappings(realmId, datasetId, dataset.getDatasetId(), upstream)
+ : List.of()))
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OpenLineage: Add Persistence schema and JDBC models [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #4827:
URL: https://github.com/apache/polaris/pull/4827#discussion_r3487974125
##########
extensions/lineage/src/main/java/org/apache/polaris/extensions/lineage/DefaultPolarisLineageHandler.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.lineage;
+
+import jakarta.enterprise.context.RequestScoped;
+import jakarta.inject.Inject;
+import java.time.Instant;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.context.CallContext;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.lineage.LineageGraph;
+import org.apache.polaris.core.lineage.LineageIngestRequest;
+import org.apache.polaris.core.lineage.LineagePersistence;
+import org.apache.polaris.core.lineage.LineageQueryRequest;
+import org.apache.polaris.core.lineage.LineageService;
+
+@RequestScoped
+public class DefaultLineageService implements LineageService {
+ private final CallContext callContext;
+ private final LineageConfiguration configuration;
+ private final LineagePersistence persistence;
+
+ @Inject
+ public DefaultLineageService(
+ CallContext callContext, LineageConfiguration configuration, LineagePersistence persistence) {
+ this.callContext = callContext;
+ this.configuration = configuration;
+ this.persistence = persistence;
+ }
Review Comment:
fixed
##########
persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java:
##########
@@ -1354,4 +1392,500 @@ private void writeCommitMetricsReport(@NonNull ModelCommitMetricsReport report)
String.format("Failed to write commit metrics report due to %s", e.getMessage()), e);
}
}
+
+ // ============================================================================
+ // LineagePersistence Implementation
+ // ============================================================================
+
+ @Override
+ public void upsertDatasets(RealmContext realmContext, List<LineageDataset> datasets) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long nowMillis = Instant.now().toEpochMilli();
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OpenLineage: Add Persistence schema and JDBC models [polaris]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4827:
URL: https://github.com/apache/polaris/pull/4827#discussion_r3449811649
##########
runtime/service/src/main/java/org/apache/polaris/service/lineage/DefaultLineageService.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.lineage;
+
+import jakarta.enterprise.context.RequestScoped;
+import jakarta.inject.Inject;
+import java.time.Instant;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.context.CallContext;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.lineage.LineageGraph;
+import org.apache.polaris.core.lineage.LineageIngestRequest;
+import org.apache.polaris.core.lineage.LineagePersistence;
+import org.apache.polaris.core.lineage.LineageQueryRequest;
+import org.apache.polaris.core.lineage.LineageService;
+
+@RequestScoped
+public class DefaultLineageService implements LineageService {
+ private final CallContext callContext;
+ private final LineageConfiguration configuration;
+ private final LineagePersistence persistence;
+
+ @Inject
+ public DefaultLineageService(
+ CallContext callContext, LineageConfiguration configuration, LineagePersistence persistence) {
+ this.callContext = callContext;
+ this.configuration = configuration;
+ this.persistence = persistence;
+ }
Review Comment:
`DefaultLineageService` injects `LineagePersistence`, but the only CDI bean in this PR appears to be `DisabledLineagePersistence` (`@DefaultBean`). Without a producer that adapts the per-request `BasePersistence` session (from `MetaStoreManagerFactory.getOrCreateSession(...)`) to `LineagePersistence`, lineage ingest/query will keep throwing even when JDBC supports it. Also, `LineageConfiguration.persistence().enabled/type` are currently unused here, so the documented config knobs won’t have any effect.
Please wire `LineagePersistence` similarly to how `MetricsPersistence` is wired in `ServiceProducers` (reuse the session when it implements the SPI), and gate usage on the intended config flag(s).
##########
persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java:
##########
@@ -1354,4 +1392,500 @@ private void writeCommitMetricsReport(@NonNull ModelCommitMetricsReport report)
String.format("Failed to write commit metrics report due to %s", e.getMessage()), e);
}
}
+
+ // ============================================================================
+ // LineagePersistence Implementation
+ // ============================================================================
+
+ @Override
+ public void upsertDatasets(RealmContext realmContext, List<LineageDataset> datasets) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long nowMillis = Instant.now().toEpochMilli();
Review Comment:
These lineage methods derive the `realmId` from the *passed* `RealmContext`. `JdbcBasePersistenceImpl` instances are created per-realm (constructor already receives `realmId`), and most other persistence operations use the instance field. Using the caller-provided realm here makes it easy for an internal misuse to write/query the wrong tenant with a session created for another realm.
Consider using `this.realmId` (and optionally asserting it matches `realmContext.getRealmIdentifier()`) for consistency and to harden tenant isolation.
##########
persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java:
##########
@@ -1354,4 +1392,500 @@ private void writeCommitMetricsReport(@NonNull ModelCommitMetricsReport report)
String.format("Failed to write commit metrics report due to %s", e.getMessage()), e);
}
}
+
+ // ============================================================================
+ // LineagePersistence Implementation
+ // ============================================================================
+
+ @Override
+ public void upsertDatasets(RealmContext realmContext, List<LineageDataset> datasets) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long nowMillis = Instant.now().toEpochMilli();
+ for (LineageDataset dataset : datasets) {
+ long datasetId =
+ lookupLineageDataset(realmId, dataset)
+ .map(ModelLineageDataset::getDatasetId)
+ .orElseGet(IdGenerator.getIdGenerator()::nextId);
+ ModelLineageDataset model =
+ ModelLineageDataset.fromDataset(dataset, realmId, datasetId, nowMillis);
+ try {
+ datasourceOperations.executeUpdate(generateLineageDatasetUpsert(model));
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to upsert lineage dataset due to %s", e.getMessage()), e);
+ }
+ }
+ }
+
+ @Override
+ public void replaceDatasetEdges(
+ RealmContext realmContext,
+ List<LineageDataset> targetDatasets,
+ List<LineageEdge> edges,
+ Instant lastEventAt) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long lastEventAtMillis = lastEventAt.toEpochMilli();
+ Map<Long, List<ModelLineageEdge>> edgesByTargetDatasetId = new LinkedHashMap<>();
+ for (LineageDataset targetDataset : targetDatasets) {
+ edgesByTargetDatasetId.putIfAbsent(
+ requireLineageDatasetId(realmId, targetDataset), new ArrayList<>());
+ }
+ for (LineageEdge edge : edges) {
+ ModelLineageEdge model =
+ ModelLineageEdge.fromIds(
+ realmId,
+ requireLineageDatasetId(realmId, edge.source()),
+ requireLineageDatasetId(realmId, edge.target()),
+ lastEventAtMillis);
+ edgesByTargetDatasetId
+ .computeIfAbsent(model.getTargetDatasetId(), ignored -> new ArrayList<>())
+ .add(model);
+ }
+
+ try {
+ datasourceOperations.runWithinTransaction(
+ connection -> {
+ for (Map.Entry<Long, List<ModelLineageEdge>> entry :
+ edgesByTargetDatasetId.entrySet()) {
+ long targetDatasetId = entry.getKey();
+ Optional<Long> currentLastEventAt =
+ loadLineageDatasetLastEventAt(connection, realmId, targetDatasetId);
+ if (currentLastEventAt.isPresent() && currentLastEventAt.get() > lastEventAtMillis) {
+ continue;
+ }
+
+ List<Long> sourceDatasetIds =
+ entry.getValue().stream()
+ .map(ModelLineageEdge::getSourceDatasetId)
+ .distinct()
+ .toList();
+ datasourceOperations.execute(
+ connection,
+ generateDeleteStaleLineageEdges(realmId, targetDatasetId, sourceDatasetIds));
+ datasourceOperations.execute(
+ connection,
+ generateDeleteStaleLineageColumnEdges(
+ realmId, targetDatasetId, sourceDatasetIds));
+ for (ModelLineageEdge model : entry.getValue()) {
+ datasourceOperations.execute(connection, generateLineageEdgeUpsert(model));
+ }
+ datasourceOperations.execute(
+ connection,
+ generateLineageDatasetLastEventAtUpdate(
+ realmId, targetDatasetId, lastEventAtMillis));
+ }
+ return true;
+ });
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to upsert lineage edge due to %s", e.getMessage()), e);
+ }
+ }
+
+ private Optional<Long> loadLineageDatasetLastEventAt(
+ Connection connection, String realmId, long targetDatasetId) throws SQLException {
+ String table = QueryGenerator.getFullyQualifiedTableName(ModelLineageDataset.TABLE_NAME);
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "SELECT last_lineage_event_at FROM "
+ + table
+ + " WHERE realm_id = ? AND dataset_id = ?")) {
+ statement.setString(1, realmId);
+ statement.setLong(2, targetDatasetId);
+ try (ResultSet resultSet = statement.executeQuery()) {
+ if (!resultSet.next()) {
+ return Optional.empty();
+ }
+ long value = resultSet.getLong(1);
+ return resultSet.wasNull() ? Optional.empty() : Optional.of(value);
+ }
+ }
+ }
+
+ @Override
+ public void upsertColumnEdges(
+ RealmContext realmContext, List<LineageColumnEdge> columnEdges, Instant lastEventAt) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long lastEventAtMillis = lastEventAt.toEpochMilli();
+ for (LineageColumnEdge columnEdge : columnEdges) {
+ ModelLineageDataset targetDataset =
+ requireLineageDataset(realmId, columnEdge.target().dataset());
+ Long currentLastEventAt = targetDataset.getLastLineageEventAt();
+ if (currentLastEventAt != null && currentLastEventAt > lastEventAtMillis) {
+ continue;
+ }
+
+ ModelLineageColumnEdge model =
+ ModelLineageColumnEdge.fromIds(
+ realmId,
+ requireLineageDatasetId(realmId, columnEdge.source().dataset()),
+ columnEdge.source().field(),
+ targetDataset.getDatasetId(),
+ columnEdge.target().field(),
+ lastEventAtMillis);
+ try {
+ datasourceOperations.executeUpdate(generateLineageColumnEdgeUpsert(model));
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to upsert lineage column edge due to %s", e.getMessage()), e);
+ }
+ }
+ }
+
+ @Override
+ public LineageGraph loadLineage(RealmContext realmContext, LineageQueryRequest request) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ Optional<ModelLineageDataset> requested =
+ lookupLineageDatasetByNodeId(realmId, request.nodeId());
+ if (requested.isEmpty()) {
+ return new LineageGraph(
+ new LineageNode(request.nodeId(), LineageNodeType.DATASET, null, true),
+ List.of(),
+ List.of());
+ }
+
+ ModelLineageDataset dataset = requested.get();
+ boolean includeColumns = request.granularity() == LineageGranularity.COLUMN;
+ List<LineageNode> upstream =
+ request.direction() == LineageDirection.UPSTREAM
+ || request.direction() == LineageDirection.BOTH
+ ? loadAdjacentLineageNodes(realmId, dataset.getDatasetId(), true, includeColumns)
+ : List.of();
+ List<LineageNode> downstream =
+ request.direction() == LineageDirection.DOWNSTREAM
+ || request.direction() == LineageDirection.BOTH
+ ? loadAdjacentLineageNodes(realmId, dataset.getDatasetId(), false, includeColumns)
+ : List.of();
+
+ return new LineageGraph(toLineageNode(dataset, List.of()), upstream, downstream);
+ }
+
+ private Optional<ModelLineageDataset> lookupLineageDataset(
+ String realmId, LineageDataset dataset) {
+ String table = QueryGenerator.getFullyQualifiedTableName(ModelLineageDataset.TABLE_NAME);
+ PreparedQuery query =
+ new PreparedQuery(
+ "SELECT realm_id, dataset_id, catalog, namespace, name, polaris_entity_id, last_lineage_event_at, created_at, updated_at "
+ + "FROM "
+ + table
+ + " WHERE realm_id = ? AND namespace = ? AND name = ?",
+ List.of(realmId, dataset.namespace(), dataset.name()));
+ try {
+ List<ModelLineageDataset> results =
+ datasourceOperations.executeSelect(query, ModelLineageDataset.CONVERTER);
+ return results.stream().findFirst();
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to load lineage dataset due to %s", e.getMessage()), e);
+ }
+ }
+
+ private Optional<ModelLineageDataset> lookupLineageDatasetByNodeId(
+ String realmId, String nodeId) {
+ return parseLineageDatasetNodeId(nodeId)
+ .flatMap(dataset -> lookupLineageDataset(realmId, dataset));
+ }
+
+ private Optional<LineageDataset> parseLineageDatasetNodeId(String nodeId) {
+ String prefix = "dataset:";
+ if (!nodeId.startsWith(prefix)) {
+ return Optional.empty();
+ }
+ String identity = nodeId.substring(prefix.length());
+ int catalogSeparator = identity.lastIndexOf(':');
+ int nameSeparator = identity.lastIndexOf('.');
+ if (catalogSeparator < 0 || nameSeparator <= catalogSeparator + 1) {
+ return Optional.empty();
+ }
+ String catalog = identity.substring(0, catalogSeparator);
+ String namespace = identity.substring(catalogSeparator + 1, nameSeparator);
+ String name = identity.substring(nameSeparator + 1);
+ return Optional.of(new LineageDataset(catalog, namespace, name));
+ }
+
+ private long requireLineageDatasetId(String realmId, LineageDataset dataset) {
+ return requireLineageDataset(realmId, dataset).getDatasetId();
+ }
+
+ private ModelLineageDataset requireLineageDataset(String realmId, LineageDataset dataset) {
+ return lookupLineageDataset(realmId, dataset)
+ .orElseThrow(
+ () ->
+ new IllegalArgumentException(
+ String.format(
+ "Lineage dataset '%s.%s' does not exist in realm '%s'. Call upsertDatasets before writing edges.",
+ dataset.namespace(), dataset.name(), realmId)));
+ }
+
+ private List<LineageNode> loadAdjacentLineageNodes(
+ String realmId, long datasetId, boolean upstream, boolean includeColumns) {
+ String edgesTable = QueryGenerator.getFullyQualifiedTableName(ModelLineageEdge.TABLE_NAME);
+ String datasetsTable =
+ QueryGenerator.getFullyQualifiedTableName(ModelLineageDataset.TABLE_NAME);
+ String adjacentEdgeColumn = upstream ? "source_dataset_id" : "target_dataset_id";
+ String requestedEdgeColumn = upstream ? "target_dataset_id" : "source_dataset_id";
+ PreparedQuery query =
+ new PreparedQuery(
+ "SELECT d.realm_id, d.dataset_id, d.catalog, d.namespace, d.name, d.polaris_entity_id, d.last_lineage_event_at, d.created_at, d.updated_at "
+ + "FROM "
+ + edgesTable
+ + " e JOIN "
+ + datasetsTable
+ + " d ON d.realm_id = e.realm_id AND d.dataset_id = e."
+ + adjacentEdgeColumn
+ + " WHERE e.realm_id = ? AND e."
+ + requestedEdgeColumn
+ + " = ?",
+ List.of(realmId, datasetId));
+ try {
+ List<ModelLineageDataset> datasets =
+ datasourceOperations.executeSelect(query, ModelLineageDataset.CONVERTER);
+ return datasets.stream()
+ .map(
+ dataset ->
+ toLineageNode(
+ dataset,
+ includeColumns
+ ? loadFieldMappings(realmId, datasetId, dataset.getDatasetId(), upstream)
+ : List.of()))
Review Comment:
With `includeColumns` enabled, this stream mapping calls `loadFieldMappings(...)` once per adjacent dataset (N+1 query pattern). For nodes with many direct upstream/downstream neighbors, a single lineage request can fan out into many extra DB queries.
Consider batching field-mapping loads (single query for all adjacent dataset ids + grouping in-memory) to keep query count bounded.
##########
persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java:
##########
@@ -1354,4 +1392,500 @@ private void writeCommitMetricsReport(@NonNull ModelCommitMetricsReport report)
String.format("Failed to write commit metrics report due to %s", e.getMessage()), e);
}
}
+
+ // ============================================================================
+ // LineagePersistence Implementation
+ // ============================================================================
+
+ @Override
+ public void upsertDatasets(RealmContext realmContext, List<LineageDataset> datasets) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long nowMillis = Instant.now().toEpochMilli();
+ for (LineageDataset dataset : datasets) {
+ long datasetId =
+ lookupLineageDataset(realmId, dataset)
+ .map(ModelLineageDataset::getDatasetId)
+ .orElseGet(IdGenerator.getIdGenerator()::nextId);
+ ModelLineageDataset model =
+ ModelLineageDataset.fromDataset(dataset, realmId, datasetId, nowMillis);
+ try {
+ datasourceOperations.executeUpdate(generateLineageDatasetUpsert(model));
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to upsert lineage dataset due to %s", e.getMessage()), e);
+ }
+ }
+ }
+
+ @Override
+ public void replaceDatasetEdges(
+ RealmContext realmContext,
+ List<LineageDataset> targetDatasets,
+ List<LineageEdge> edges,
+ Instant lastEventAt) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long lastEventAtMillis = lastEventAt.toEpochMilli();
+ Map<Long, List<ModelLineageEdge>> edgesByTargetDatasetId = new LinkedHashMap<>();
+ for (LineageDataset targetDataset : targetDatasets) {
+ edgesByTargetDatasetId.putIfAbsent(
+ requireLineageDatasetId(realmId, targetDataset), new ArrayList<>());
+ }
+ for (LineageEdge edge : edges) {
+ ModelLineageEdge model =
+ ModelLineageEdge.fromIds(
+ realmId,
+ requireLineageDatasetId(realmId, edge.source()),
+ requireLineageDatasetId(realmId, edge.target()),
+ lastEventAtMillis);
+ edgesByTargetDatasetId
+ .computeIfAbsent(model.getTargetDatasetId(), ignored -> new ArrayList<>())
+ .add(model);
+ }
+
+ try {
+ datasourceOperations.runWithinTransaction(
+ connection -> {
+ for (Map.Entry<Long, List<ModelLineageEdge>> entry :
+ edgesByTargetDatasetId.entrySet()) {
+ long targetDatasetId = entry.getKey();
+ Optional<Long> currentLastEventAt =
+ loadLineageDatasetLastEventAt(connection, realmId, targetDatasetId);
+ if (currentLastEventAt.isPresent() && currentLastEventAt.get() > lastEventAtMillis) {
+ continue;
+ }
+
+ List<Long> sourceDatasetIds =
+ entry.getValue().stream()
+ .map(ModelLineageEdge::getSourceDatasetId)
+ .distinct()
+ .toList();
+ datasourceOperations.execute(
+ connection,
+ generateDeleteStaleLineageEdges(realmId, targetDatasetId, sourceDatasetIds));
+ datasourceOperations.execute(
+ connection,
+ generateDeleteStaleLineageColumnEdges(
+ realmId, targetDatasetId, sourceDatasetIds));
+ for (ModelLineageEdge model : entry.getValue()) {
+ datasourceOperations.execute(connection, generateLineageEdgeUpsert(model));
+ }
+ datasourceOperations.execute(
+ connection,
+ generateLineageDatasetLastEventAtUpdate(
+ realmId, targetDatasetId, lastEventAtMillis));
+ }
+ return true;
+ });
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to upsert lineage edge due to %s", e.getMessage()), e);
+ }
+ }
+
+ private Optional<Long> loadLineageDatasetLastEventAt(
+ Connection connection, String realmId, long targetDatasetId) throws SQLException {
+ String table = QueryGenerator.getFullyQualifiedTableName(ModelLineageDataset.TABLE_NAME);
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "SELECT last_lineage_event_at FROM "
+ + table
+ + " WHERE realm_id = ? AND dataset_id = ?")) {
+ statement.setString(1, realmId);
+ statement.setLong(2, targetDatasetId);
+ try (ResultSet resultSet = statement.executeQuery()) {
+ if (!resultSet.next()) {
+ return Optional.empty();
+ }
+ long value = resultSet.getLong(1);
+ return resultSet.wasNull() ? Optional.empty() : Optional.of(value);
+ }
+ }
+ }
+
+ @Override
+ public void upsertColumnEdges(
+ RealmContext realmContext, List<LineageColumnEdge> columnEdges, Instant lastEventAt) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ long lastEventAtMillis = lastEventAt.toEpochMilli();
+ for (LineageColumnEdge columnEdge : columnEdges) {
+ ModelLineageDataset targetDataset =
+ requireLineageDataset(realmId, columnEdge.target().dataset());
+ Long currentLastEventAt = targetDataset.getLastLineageEventAt();
+ if (currentLastEventAt != null && currentLastEventAt > lastEventAtMillis) {
+ continue;
+ }
+
+ ModelLineageColumnEdge model =
+ ModelLineageColumnEdge.fromIds(
+ realmId,
+ requireLineageDatasetId(realmId, columnEdge.source().dataset()),
+ columnEdge.source().field(),
+ targetDataset.getDatasetId(),
+ columnEdge.target().field(),
+ lastEventAtMillis);
+ try {
+ datasourceOperations.executeUpdate(generateLineageColumnEdgeUpsert(model));
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ String.format("Failed to upsert lineage column edge due to %s", e.getMessage()), e);
+ }
+ }
+ }
+
+ @Override
+ public LineageGraph loadLineage(RealmContext realmContext, LineageQueryRequest request) {
+ verifyLineagePersistenceSupported();
+ String realmId = realmContext.getRealmIdentifier();
+ Optional<ModelLineageDataset> requested =
+ lookupLineageDatasetByNodeId(realmId, request.nodeId());
+ if (requested.isEmpty()) {
+ return new LineageGraph(
+ new LineageNode(request.nodeId(), LineageNodeType.DATASET, null, true),
+ List.of(),
+ List.of());
+ }
+
+ ModelLineageDataset dataset = requested.get();
+ boolean includeColumns = request.granularity() == LineageGranularity.COLUMN;
+ List<LineageNode> upstream =
+ request.direction() == LineageDirection.UPSTREAM
+ || request.direction() == LineageDirection.BOTH
+ ? loadAdjacentLineageNodes(realmId, dataset.getDatasetId(), true, includeColumns)
+ : List.of();
+ List<LineageNode> downstream =
+ request.direction() == LineageDirection.DOWNSTREAM
+ || request.direction() == LineageDirection.BOTH
+ ? loadAdjacentLineageNodes(realmId, dataset.getDatasetId(), false, includeColumns)
+ : List.of();
+
+ return new LineageGraph(toLineageNode(dataset, List.of()), upstream, downstream);
+ }
+
+ private Optional<ModelLineageDataset> lookupLineageDataset(
+ String realmId, LineageDataset dataset) {
+ String table = QueryGenerator.getFullyQualifiedTableName(ModelLineageDataset.TABLE_NAME);
+ PreparedQuery query =
+ new PreparedQuery(
+ "SELECT realm_id, dataset_id, catalog, namespace, name, polaris_entity_id, last_lineage_event_at, created_at, updated_at "
+ + "FROM "
+ + table
+ + " WHERE realm_id = ? AND namespace = ? AND name = ?",
+ List.of(realmId, dataset.namespace(), dataset.name()));
Review Comment:
`lookupLineageDataset(...)` (and the table uniqueness in schema-v5) identify datasets by `(realm_id, namespace, name)` only, but the public node id format includes `catalog` (`dataset:<catalog>:<namespace>.<name>`). This makes node IDs effectively non-canonical (queries with the “wrong” catalog will still resolve), and it prevents representing two datasets that differ only by catalog/platform.
Please align identity across: (1) node-id parsing/formatting, (2) uniqueness/indexes in `lineage_datasets`, and (3) all lookup/upsert predicates (either include `catalog` everywhere, or remove it from the node id and schema).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OpenLineage: Add Persistence schema and JDBC models [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4827:
URL: https://github.com/apache/polaris/pull/4827#discussion_r3444250728
##########
persistence/relational-jdbc/src/main/resources/postgres/schema-v5.sql:
##########
@@ -0,0 +1,344 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file--
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"). You may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Changes from v4:
+-- * Added lineage graph tables
+
+CREATE SCHEMA IF NOT EXISTS POLARIS_SCHEMA;
+SET search_path TO POLARIS_SCHEMA;
+
+CREATE TABLE IF NOT EXISTS version (
+ version_key TEXT PRIMARY KEY,
+ version_value INTEGER NOT NULL
+);
+INSERT INTO version (version_key, version_value)
+VALUES ('version', 5)
+ON CONFLICT (version_key) DO UPDATE
+SET version_value = EXCLUDED.version_value;
+COMMENT ON TABLE version IS 'the version of the JDBC schema in use';
+
+CREATE TABLE IF NOT EXISTS entities (
+ realm_id TEXT NOT NULL,
+ catalog_id BIGINT NOT NULL,
+ id BIGINT NOT NULL,
+ parent_id BIGINT NOT NULL,
+ name TEXT NOT NULL,
+ entity_version INT NOT NULL,
+ type_code INT NOT NULL,
+ sub_type_code INT NOT NULL,
+ create_timestamp BIGINT NOT NULL,
+ drop_timestamp BIGINT NOT NULL,
+ purge_timestamp BIGINT NOT NULL,
+ to_purge_timestamp BIGINT NOT NULL,
+ last_update_timestamp BIGINT NOT NULL,
+ properties JSONB not null default '{}'::JSONB,
+ internal_properties JSONB not null default '{}'::JSONB,
+ grant_records_version INT NOT NULL,
+ location_without_scheme TEXT,
+ PRIMARY KEY (realm_id, id),
+ CONSTRAINT constraint_name UNIQUE (realm_id, catalog_id, parent_id, type_code, name)
+);
+
+-- TODO: create indexes based on all query pattern.
+CREATE INDEX IF NOT EXISTS idx_entities ON entities (realm_id, catalog_id, id);
+CREATE INDEX IF NOT EXISTS idx_entities_catalog_id_id ON entities (catalog_id, id);
+CREATE INDEX IF NOT EXISTS idx_locations
+ ON entities USING btree (realm_id, parent_id, location_without_scheme)
+ WHERE location_without_scheme IS NOT NULL;
+
+COMMENT ON TABLE entities IS 'all the entities';
+
+COMMENT ON COLUMN entities.realm_id IS 'realm_id used for multi-tenancy';
+COMMENT ON COLUMN entities.catalog_id IS 'catalog id';
+COMMENT ON COLUMN entities.id IS 'entity id';
+COMMENT ON COLUMN entities.parent_id IS 'entity id of parent';
+COMMENT ON COLUMN entities.name IS 'entity name';
+COMMENT ON COLUMN entities.entity_version IS 'version of the entity';
+COMMENT ON COLUMN entities.type_code IS 'type code';
+COMMENT ON COLUMN entities.sub_type_code IS 'sub type of entity';
+COMMENT ON COLUMN entities.create_timestamp IS 'creation time of entity';
+COMMENT ON COLUMN entities.drop_timestamp IS 'time of drop of entity';
+COMMENT ON COLUMN entities.purge_timestamp IS 'time to start purging entity';
+COMMENT ON COLUMN entities.last_update_timestamp IS 'last time the entity is touched';
+COMMENT ON COLUMN entities.properties IS 'entities properties json';
+COMMENT ON COLUMN entities.internal_properties IS 'entities internal properties json';
+COMMENT ON COLUMN entities.grant_records_version IS 'the version of grant records change on the entity';
+
+CREATE TABLE IF NOT EXISTS grant_records (
+ realm_id TEXT NOT NULL,
+ securable_catalog_id BIGINT NOT NULL,
+ securable_id BIGINT NOT NULL,
+ grantee_catalog_id BIGINT NOT NULL,
+ grantee_id BIGINT NOT NULL,
+ privilege_code INTEGER,
+ PRIMARY KEY (realm_id, securable_catalog_id, securable_id, grantee_catalog_id, grantee_id, privilege_code)
+);
+
+COMMENT ON TABLE grant_records IS 'grant records for entities';
+
+COMMENT ON COLUMN grant_records.securable_catalog_id IS 'catalog id of the securable';
+COMMENT ON COLUMN grant_records.securable_id IS 'entity id of the securable';
+COMMENT ON COLUMN grant_records.grantee_catalog_id IS 'catalog id of the grantee';
+COMMENT ON COLUMN grant_records.grantee_id IS 'id of the grantee';
+COMMENT ON COLUMN grant_records.privilege_code IS 'privilege code';
+
+CREATE INDEX IF NOT EXISTS idx_grants_realm_grantee
+ ON grant_records (realm_id, grantee_id);
+CREATE INDEX IF NOT EXISTS idx_grants_realm_securable
+ ON grant_records (realm_id, securable_id);
+
+CREATE TABLE IF NOT EXISTS principal_authentication_data (
+ realm_id TEXT NOT NULL,
+ principal_id BIGINT NOT NULL,
+ principal_client_id VARCHAR(255) NOT NULL,
+ main_secret_hash VARCHAR(255) NOT NULL,
+ secondary_secret_hash VARCHAR(255) NOT NULL,
+ secret_salt VARCHAR(255) NOT NULL,
+ PRIMARY KEY (realm_id, principal_client_id)
+);
+
+COMMENT ON TABLE principal_authentication_data IS 'authentication data for client';
+
+CREATE TABLE IF NOT EXISTS policy_mapping_record (
+ realm_id TEXT NOT NULL,
+ target_catalog_id BIGINT NOT NULL,
+ target_id BIGINT NOT NULL,
+ policy_type_code INTEGER NOT NULL,
+ policy_catalog_id BIGINT NOT NULL,
+ policy_id BIGINT NOT NULL,
+ parameters JSONB NOT NULL DEFAULT '{}'::JSONB,
+ PRIMARY KEY (realm_id, target_catalog_id, target_id, policy_type_code, policy_catalog_id, policy_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_policy_mapping_record ON policy_mapping_record (realm_id, policy_type_code, policy_catalog_id, policy_id, target_catalog_id, target_id);
+
+CREATE TABLE IF NOT EXISTS events (
+ realm_id TEXT NOT NULL,
+ catalog_id TEXT NOT NULL,
+ event_id TEXT NOT NULL,
+ request_id TEXT,
+ event_type TEXT NOT NULL,
+ timestamp_ms BIGINT NOT NULL,
+ principal_name TEXT,
+ resource_type TEXT NOT NULL,
+ resource_identifier TEXT NOT NULL,
+ additional_properties JSONB NOT NULL DEFAULT '{}'::JSONB,
+ PRIMARY KEY (event_id)
+);
+
+-- Idempotency records (key-only idempotency; durable replay)
+CREATE TABLE IF NOT EXISTS idempotency_records (
+ realm_id TEXT NOT NULL,
+ idempotency_key TEXT NOT NULL,
+ operation_type TEXT NOT NULL,
+ resource_id TEXT NOT NULL, -- normalized request-derived resource identifier (not a generated entity id)
+
+ -- Finalization/replay
+ http_status INTEGER, -- NULL while IN_PROGRESS; set only on finalized 2xx/terminal 4xx
+ error_subtype TEXT, -- optional: e.g., already_exists, namespace_not_empty, idempotency_replay_failed
+ response_summary TEXT, -- minimal body to reproduce equivalent response (JSON string)
+ response_headers TEXT, -- small whitelisted headers to replay (JSON string)
+ finalized_at TIMESTAMP, -- when http_status was written
+
+ -- Liveness/ops
+ created_at TIMESTAMP NOT NULL,
+ updated_at TIMESTAMP NOT NULL,
+ heartbeat_at TIMESTAMP, -- updated by owner while IN_PROGRESS
+ executor_id TEXT, -- owner pod/worker id
+ expires_at TIMESTAMP,
+
+ PRIMARY KEY (realm_id, idempotency_key)
+);
+
+-- Helpful indexes
+CREATE INDEX IF NOT EXISTS idx_idemp_realm_expires
+ ON idempotency_records (realm_id, expires_at);
+
+-- ============================================================================
+-- SCAN METRICS REPORT TABLE
+-- ============================================================================
+
+CREATE TABLE IF NOT EXISTS scan_metrics_report (
Review Comment:
I think our SQL DDL scripts jam too many features into the same file. I'd like to propose splitting them by feature (MetaStore, Metrics, Events, OpenLineage).
Does OL data absolutely have to be in the same RDBMS schema as the MetaStore data?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OpenLineage: Add Persistence schema and JDBC models [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on PR #4827:
URL: https://github.com/apache/polaris/pull/4827#issuecomment-5437194612
reopen
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org