Merge branch 'main' into web/add_paging_dataset_version_hist

Author: davidsharp7

api/src/main/java/marquez/api/JobResource.java

@@ -238,7 +238,8 @@ public Response listRuns(
 
     final List<Run> runs =
         runService.findAll(namespaceName.getValue(), jobName.getValue(), limit, offset);
-    return Response.ok(new Runs(runs)).build();
+    final int totalCount = jobService.countJobRuns(namespaceName.getValue(), jobName.getValue());
+    return Response.ok(new Runs(runs, totalCount)).build();
   }
 
   @Path("/jobs/runs/{id}")
@@ -329,5 +330,8 @@ public static class Runs {
     @NonNull
     @JsonProperty("runs")
     List<Run> value;
+
+    @JsonProperty("totalCount")
+    int totalCount;
   }
 }

api/src/main/java/marquez/db/DatasetVersionDao.java

@@ -278,13 +278,19 @@ LEFT JOIN (
       	WHERE dv.namespace_name = :namespaceName
             AND dv.dataset_name = :datasetName
       	LIMIT :limit OFFSET :offset
+        ),
+        dataset_symlinks_names as (
+          SELECT DISTINCT dataset_uuid, name
+          FROM dataset_symlinks
+          WHERE NOT is_primary
         )
         SELECT
 	        type, name, physical_name, namespace_name, source_name, description, lifecycle_state,
             created_at, version, dataset_schema_version_uuid, fields, createdByRunUuid, schema_location,
             tags, dataset_version_uuid,
 	        JSONB_AGG(facets ORDER BY lineage_event_time ASC) AS facets
         FROM dataset_info
+        WHERE name NOT IN (SELECT name FROM dataset_symlinks_names)
         GROUP BY type, name, physical_name, namespace_name, source_name, description, lifecycle_state,
             created_at, version, dataset_schema_version_uuid, fields, createdByRunUuid, schema_location,
             tags, dataset_version_uuid

api/src/main/java/marquez/db/JobDao.java

@@ -252,6 +252,20 @@ job_tags as (
   @SqlQuery("SELECT count(*) FROM jobs_view AS j WHERE symlink_target_uuid IS NULL")
   int count();
 
+  @SqlQuery(
+      """
+      select
+          count(*)
+      from
+          runs
+      where
+          namespace_name = :namespaceName
+      and
+          job_name = :job
+      ;
+      """)
+  int countJobRuns(String namespaceName, String job);
+
   @SqlQuery(
       "SELECT count(*) FROM jobs_view AS j WHERE j.namespace_name = :namespaceName\n"
           + "AND symlink_target_uuid IS NULL")

api/src/test/java/marquez/DatasetIntegrationTest.java

@@ -7,6 +7,8 @@
 
 import static marquez.db.ColumnLineageTestUtils.getDatasetA;
 import static marquez.db.ColumnLineageTestUtils.getDatasetB;
+import static marquez.db.LineageTestUtils.PRODUCER_URL;
+import static marquez.db.LineageTestUtils.SCHEMA_URL;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import com.fasterxml.jackson.core.type.TypeReference;
@@ -574,4 +576,57 @@ public void testApp_doesNotShowDeletedDatasetAfterUndeleteNamespace() throws IOE
     jobs = client.listJobs(namespaceName);
     assertThat(jobs).hasSize(1);
   }
+
+  @Test
+  public void testApp_getTableVersionsWithSymlinks() {
+    client.createDataset(NAMESPACE_NAME, DB_TABLE_NAME, DB_TABLE_META);
+
+    ImmutableMap<String, Object> outputFacets =
+        ImmutableMap.of("outputFacetKey", "outputFacetValue");
+    ImmutableMap<String, Object> inputFacets = ImmutableMap.of("inputFacetKey", "inputFacetValue");
+
+    final LineageEvent.DatasetFacets datasetFacets =
+        LineageTestUtils.newDatasetFacet(
+            outputFacets,
+            LineageEvent.SchemaField.builder()
+                .name("firstname")
+                .type("string")
+                .description("the first name")
+                .build());
+    datasetFacets
+        .getDocumentation()
+        .setDescription(DB_TABLE_META.getDescription().orElse("the dataset documentation"));
+    datasetFacets.setSymlinks(
+        new LineageEvent.DatasetSymlinkFacet(
+            PRODUCER_URL,
+            SCHEMA_URL,
+            Collections.singletonList(
+                new LineageEvent.SymlinkIdentifier("symlinkNamespace", "symlinkName", "type"))));
+    final LineageEvent lineageEvent =
+        LineageEvent.builder()
+            .producer("testApp_getTableVersionsWithSymlinks")
+            .eventType("COMPLETE")
+            .run(
+                new LineageEvent.Run(
+                    UUID.randomUUID().toString(), LineageEvent.RunFacet.builder().build()))
+            .job(LineageEvent.Job.builder().namespace(NAMESPACE_NAME).name(JOB_NAME).build())
+            .eventTime(ZonedDateTime.now())
+            .inputs(
+                Collections.singletonList(
+                    LineageEvent.Dataset.builder()
+                        .namespace(NAMESPACE_NAME)
+                        .name(DB_TABLE_NAME)
+                        .facets(datasetFacets)
+                        .build()))
+            .outputs(Collections.emptyList())
+            .build();
+    final CompletableFuture<Integer> resp = sendEvent(lineageEvent);
+    assertThat(resp.join()).isEqualTo(201);
+    List<DatasetVersion> versions = client.listDatasetVersions(NAMESPACE_NAME, DB_TABLE_NAME);
+
+    versions.forEach(
+        datasetVersion -> {
+          assertThat(datasetVersion.getName()).isNotEqualTo("symlinkName");
+        });
+  }
 }

api/src/test/java/marquez/db/JobDaoTest.java

@@ -136,6 +136,7 @@ public void testCountFor() {
     assertThat(jobDao.count()).isEqualTo(4);
 
     assertThat(jobDao.countFor(namespace.getName())).isEqualTo(3);
+    assertThat(jobDao.countJobRuns(namespace.getName(), "targetJob")).isEqualTo(0);
   }
 
   @Test

web/src/components/jobs/JobDetailPage.tsx

@@ -20,7 +20,7 @@ import {
   deleteJob,
   dialogToggle,
   fetchJobTags,
-  fetchRuns,
+  fetchLatestRuns,
   resetJobs,
   resetRuns,
   setTabIndex,
@@ -44,7 +44,7 @@ import Runs from './Runs'
 import SpeedRounded from '@mui/icons-material/SpeedRounded'
 
 interface DispatchProps {
-  fetchRuns: typeof fetchRuns
+  fetchLatestRuns: typeof fetchLatestRuns
   resetRuns: typeof resetRuns
   resetJobs: typeof resetJobs
   deleteJob: typeof deleteJob
@@ -56,41 +56,42 @@ interface DispatchProps {
 type IProps = {
   job: LineageJob
   jobs: IState['jobs']
-  runs: Run[]
-  runsLoading: boolean
   display: IState['display']
   tabIndex: IState['lineage']['tabIndex']
   jobTags: string[]
+  latestRuns: Run[]
+  isLatestRunsLoading: boolean
 } & DispatchProps
 
 const JobDetailPage: FunctionComponent<IProps> = (props) => {
   const theme = createTheme(useTheme())
   const {
     job,
     jobs,
-    fetchRuns,
+    fetchLatestRuns,
     resetRuns,
     deleteJob,
     dialogToggle,
-    runs,
+    latestRuns,
     display,
-    runsLoading,
     tabIndex,
     setTabIndex,
     jobTags,
     fetchJobTags,
+    isLatestRunsLoading,
   } = props
   const navigate = useNavigate()
   const [_, setSearchParams] = useSearchParams()
 
   const handleChange = (event: ChangeEvent, newValue: number) => {
     setTabIndex(newValue)
   }
+
   const i18next = require('i18next')
 
   useEffect(() => {
-    fetchRuns(job.name, job.namespace)
     fetchJobTags(job.namespace, job.name)
+    fetchLatestRuns(job.name, job.namespace)
   }, [job.name])
 
   useEffect(() => {
@@ -102,12 +103,13 @@ const JobDetailPage: FunctionComponent<IProps> = (props) => {
   // unmounting
   useEffect(() => {
     return () => {
-      resetRuns()
       resetJobs()
+      resetRuns()
+      setTabIndex(0)
     }
   }, [])
 
-  if (runsLoading || jobs.isLoading) {
+  if (jobs.isLoading || isLatestRunsLoading) {
     return (
       <Box display={'flex'} justifyContent={'center'} mt={2}>
         <CircularProgress color='primary' />
@@ -244,7 +246,7 @@ const JobDetailPage: FunctionComponent<IProps> = (props) => {
           <MqInfo
             icon={<DirectionsRun color={'disabled'} />}
             label={'Running Status'.toUpperCase()}
-            value={<MqStatus label={job.latestRun?.state} color={jobRunsStatus(runs)} />}
+            value={<MqStatus label={job.latestRun?.state} color={jobRunsStatus(latestRuns)} />}
           />
         </Grid>
       </Grid>
@@ -271,14 +273,14 @@ const JobDetailPage: FunctionComponent<IProps> = (props) => {
           )
         )
       ) : null}
-      {tabIndex === 1 && <Runs runs={runs} />}
+      {tabIndex === 1 && <Runs jobName={job.name} jobNamespace={job.namespace} />}
     </Box>
   )
 }
 
 const mapStateToProps = (state: IState) => ({
-  runs: state.runs.result,
-  runsLoading: state.runs.isLoading,
+  latestRuns: state.runs.latestRuns,
+  isLatestRunsLoading: state.runs.isLatestRunsLoading,
   display: state.display,
   jobs: state.jobs,
   tabIndex: state.lineage.tabIndex,
@@ -288,7 +290,7 @@ const mapStateToProps = (state: IState) => ({
 const mapDispatchToProps = (dispatch: Redux.Dispatch) =>
   bindActionCreators(
     {
-      fetchRuns: fetchRuns,
+      fetchLatestRuns: fetchLatestRuns,
       resetRuns: resetRuns,
       resetJobs: resetJobs,
       deleteJob: deleteJob,

web/src/components/jobs/Runs.tsx

@@ -1,50 +1,94 @@
 // Copyright 2018-2023 contributors to the Marquez project
 // SPDX-License-Identifier: Apache-2.0
 
+import * as Redux from 'redux'
 import { ArrowBackIosRounded } from '@mui/icons-material'
 import {
   Box,
   Chip,
+  CircularProgress,
   IconButton,
   Table,
   TableBody,
   TableCell,
   TableHead,
   TableRow,
 } from '@mui/material'
+import { IState } from '../../store/reducers'
 import { Run } from '../../types/api'
 import { alpha, createTheme } from '@mui/material/styles'
+import { bindActionCreators } from 'redux'
+import { connect } from 'react-redux'
+import { fetchRuns } from '../../store/actionCreators'
 import { formatUpdatedAt } from '../../helpers'
 import { runStateColor } from '../../helpers/nodes'
 import { stopWatchDuration } from '../../helpers/time'
 import { useTheme } from '@emotion/react'
 import MqCode from '../core/code/MqCode'
 import MqCopy from '../core/copy/MqCopy'
 import MqEmpty from '../core/empty/MqEmpty'
+import MqPaging from '../paging/MqPaging'
 import MqStatus from '../core/status/MqStatus'
 import MqText from '../core/text/MqText'
 import React, { FunctionComponent, SetStateAction } from 'react'
 import RunInfo from './RunInfo'
 
+interface DispatchProps {
+  fetchRuns: typeof fetchRuns
+}
+
 interface RunsProps {
   runs: Run[]
   facets?: object
+  totalCount: number
+  runsLoading: boolean
+  jobName: string
+  jobNamespace: string
+}
+
+interface RunsState {
+  page: number
 }
 
-const Runs: FunctionComponent<RunsProps> = (props) => {
-  const { runs, facets } = props
+const PAGE_SIZE = 10
+
+const Runs: FunctionComponent<RunsProps & DispatchProps> = (props) => {
+  const { runs, facets, totalCount, runsLoading, fetchRuns, jobName, jobNamespace } = props
   const i18next = require('i18next')
-  if (runs.length === 0) {
-    return <MqEmpty title={i18next.t('jobs.empty_title')} body={i18next.t('jobs.empty_body')} />
-  }
+
+  const [state, setState] = React.useState<RunsState>({
+    page: 0,
+  })
 
   const [infoView, setInfoView] = React.useState<Run | null>(null)
   const handleClick = (newValue: SetStateAction<Run | null>) => {
     setInfoView(newValue)
   }
 
+  const handleClickPage = (direction: 'prev' | 'next') => {
+    const directionPage = direction === 'next' ? state.page + 1 : state.page - 1
+    window.scrollTo(0, 0)
+    setState({ ...state, page: directionPage })
+  }
+
+  React.useEffect(() => {
+    fetchRuns(jobName, jobNamespace, PAGE_SIZE, state.page * PAGE_SIZE)
+  }, [state.page])
+
   const theme = createTheme(useTheme())
 
+  if (runs.length === 0) {
+    return <MqEmpty title={i18next.t('jobs.empty_title')} body={i18next.t('jobs.empty_body')} />
+  }
+
+  if (runsLoading) {
+    return (
+      <Box display={'flex'} justifyContent={'center'} mt={2}>
+        <CircularProgress color='primary' />
+      </Box>
+    )
+  }
+
   if (infoView) {
     return (
       <>
@@ -148,12 +192,23 @@ const Runs: FunctionComponent<RunsProps> = (props) => {
                 <TableCell align='left'>{formatUpdatedAt(run.createdAt)}</TableCell>
                 <TableCell align='left'>{formatUpdatedAt(run.startedAt)}</TableCell>
                 <TableCell align='left'>N/A</TableCell>
-                <TableCell align='left'>{stopWatchDuration(run.durationMs)}</TableCell>
+                <TableCell align='left'>
+                  {run.state === 'RUNNING' || run.state === 'NEW'
+                    ? 'N/A'
+                    : stopWatchDuration(run.durationMs)}{' '}
+                </TableCell>
               </TableRow>
             )
           })}
         </TableBody>
       </Table>
+      <MqPaging
+        pageSize={PAGE_SIZE}
+        currentPage={state.page}
+        totalCount={totalCount}
+        incrementPage={() => handleClickPage('next')}
+        decrementPage={() => handleClickPage('prev')}
+      />
       {facets && (
         <Box mt={2}>
           <Box mb={1}>
@@ -166,4 +221,18 @@ const Runs: FunctionComponent<RunsProps> = (props) => {
   )
 }
 
-export default Runs
+const mapStateToProps = (state: IState) => ({
+  runs: state.runs.result,
+  totalCount: state.runs.totalCount,
+  runsLoading: state.runs.isLoading,
+})
+
+const mapDispatchToProps = (dispatch: Redux.Dispatch) =>
+  bindActionCreators(
+    {
+      fetchRuns: fetchRuns,
+    },
+    dispatch
+  )
+
+export default connect(mapStateToProps, mapDispatchToProps)(Runs)