Skip to content

[FIX] Dataset query to get only the latest facet for each version #2859

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Jul 25, 2024
Merged

[FIX] Dataset query to get only the latest facet for each version #2859

Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
497846c
fix dataset query to get only the latest facet for each version
sophiely Jul 25, 2024
2e76688
small fix in query
sophiely Jul 25, 2024
75160fe
Merge branch 'main' into feat/fix-find-all-dataset-query
wslulciuc Jul 25, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 45 additions & 24 deletions api/src/main/java/marquez/db/DatasetDao.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,30 +121,51 @@ default void setFields(Dataset ds) {

@SqlQuery(
"""
SELECT d.*, dv.fields, dv.lifecycle_state, sv.schema_location, t.tags, facets
FROM datasets_view d
LEFT JOIN dataset_versions dv ON d.current_version_uuid = dv.uuid
LEFT JOIN stream_versions AS sv ON sv.dataset_version_uuid = dv.uuid
LEFT JOIN (
SELECT ARRAY_AGG(t.name) AS tags, m.dataset_uuid
FROM tags AS t
INNER JOIN datasets_tag_mapping AS m ON m.tag_uuid = t.uuid
GROUP BY m.dataset_uuid
) t ON t.dataset_uuid = d.uuid
LEFT JOIN (
SELECT
df.dataset_version_uuid,
JSONB_AGG(df.facet ORDER BY df.lineage_event_time ASC) AS facets
FROM dataset_facets AS df
WHERE df.facet IS NOT NULL AND
(df.type ILIKE 'dataset' OR df.type ILIKE 'unknown' OR df.type ILIKE 'input') AND
df.dataset_uuid IN (SELECT uuid FROM datasets_view WHERE namespace_name = :namespaceName ORDER BY name LIMIT :limit OFFSET :offset)
GROUP BY df.dataset_version_uuid
) f ON f.dataset_version_uuid = d.current_version_uuid
WHERE d.namespace_name = :namespaceName
ORDER BY d.name
LIMIT :limit OFFSET :offset
""")
WITH facets_t AS
(SELECT df.dataset_version_uuid,
df.facet,
df."name",
df.created_at,
rank() OVER (PARTITION BY df.dataset_version_uuid, "name"
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

❤️

ORDER BY created_at DESC) AS r
FROM dataset_facets AS df
WHERE df.facet IS NOT NULL
AND (df.type ILIKE 'dataset'
OR df.type ILIKE 'unknown'
OR df.type ILIKE 'input')
AND df.dataset_uuid IN
(SELECT UUID
FROM datasets_view
WHERE namespace_name = :namespaceName
ORDER BY name
LIMIT 10
OFFSET :offset))
SELECT d.*,
dv.fields,
dv.lifecycle_state,
sv.schema_location,
t.tags,
facets
FROM datasets_view d
LEFT JOIN dataset_versions dv ON d.current_version_uuid = dv.uuid
LEFT JOIN stream_versions AS sv ON sv.dataset_version_uuid = dv.uuid
LEFT JOIN
(SELECT ARRAY_AGG(t.name) AS tags,
m.dataset_uuid
FROM tags AS t
INNER JOIN datasets_tag_mapping AS m ON m.tag_uuid = t.uuid
GROUP BY m.dataset_uuid) t ON t.dataset_uuid = d.uuid
LEFT JOIN
(SELECT df.dataset_version_uuid,
JSONB_AGG(df.facet) AS facets
FROM facets_t AS df
WHERE r = 1
GROUP BY df.dataset_version_uuid) f ON f.dataset_version_uuid = d.current_version_uuid
WHERE d.namespace_name = :namespaceName
ORDER BY d.name
LIMIT :limit
OFFSET :offset
""")
List<Dataset> findAll(String namespaceName, int limit, int offset);

@SqlQuery("SELECT count(*) FROM datasets_view")
Expand Down