Table controller supports paging and filtering table names, and returns the most recent table data quality status.

Author: piotrczarnas

dqops/src/main/java/com/dqops/core/filesystem/localfiles/LocalFileStorageServiceImpl.java

@@ -34,6 +34,7 @@
 import java.nio.file.Path;
 import java.nio.file.attribute.FileTime;
 import java.time.Instant;
+import java.util.Comparator;
 import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -295,6 +296,7 @@ public List<HomeFolderPath> listFoldersDirect(HomeFolderPath folderPath, Path ab
             try (Stream<Path> fileList = Files.list(absolutePath)) {
                 List<HomeFolderPath> folders = fileList
                         .filter(path -> Files.isDirectory(path))
+                        .sorted()
                         .map(path -> {
                             String fileSystemName = path.getFileName().toString();
                             FolderName subFolder = FolderName.fromFileSystemName(fileSystemName);
@@ -339,6 +341,7 @@ public List<HomeFilePath> listFilesDirect(HomeFolderPath folderPath, Path absolu
             try (Stream<Path> pathStream = Files.list(absolutePath)) {
                 List<HomeFilePath> filePathsList = pathStream
                         .filter(path -> !Files.isDirectory(path))
+                        .sorted()
                         .map(path -> {
                             String fileSystemName = path.getFileName().toString();
                             return folderPath.resolveFile(fileSystemName);

dqops/src/main/java/com/dqops/data/checkresults/models/currentstatus/ColumnCurrentDataQualityStatusModel.java

@@ -35,6 +35,7 @@
 import java.time.ZoneId;
 import java.time.temporal.ChronoUnit;
 import java.util.*;
+import java.util.function.Predicate;
 
 /**
  * The column validity status. It is a summary of the results of the most recently executed data quality checks on the column.
@@ -189,10 +190,10 @@ protected ColumnCurrentDataQualityStatusModel clone() {
     /**
      * Creates a deep clone of the table status model, preserving only the checks for an expected check type.
      * All scores and the data quality KPI is recalculated for the checks that left.
-     * @param checkType Data quality check type to copy, the results of the other check types are ignored.
+     * @param checkFilter Check filter that filters the checks that should be preserved.
      * @return A deep clone of the object with results only for that check type.
      */
-    public ColumnCurrentDataQualityStatusModel cloneFilteredByCheckType(CheckType checkType) {
+    public ColumnCurrentDataQualityStatusModel cloneFilteredByCheckType(Predicate<CheckCurrentDataQualityStatusModel> checkFilter) {
         ColumnCurrentDataQualityStatusModel tableStatusClone = this.clone();
         tableStatusClone.currentSeverity = null;
         tableStatusClone.highestHistoricalSeverity = null;
@@ -207,7 +208,7 @@ public ColumnCurrentDataQualityStatusModel cloneFilteredByCheckType(CheckType ch
         tableStatusClone.checks = new LinkedHashMap<>();
 
         for (Map.Entry<String, CheckCurrentDataQualityStatusModel> keyValue : this.checks.entrySet()) {
-            if (keyValue.getValue().getCheckType() == checkType) {
+            if (checkFilter == null || checkFilter.test(keyValue.getValue())) {
                 tableStatusClone.checks.put(keyValue.getKey(), keyValue.getValue());
             }
         }

dqops/src/main/java/com/dqops/data/checkresults/models/currentstatus/TableCurrentDataQualityStatusModel.java

@@ -34,6 +34,8 @@
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.*;
+import java.util.function.Function;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -308,11 +310,12 @@ protected TableCurrentDataQualityStatusModel clone() {
     /**
      * Creates a deep clone of the table status model, preserving only the checks for an expected check type.
      * All scores and the data quality KPI is recalculated for the checks that left.
-     * @param checkType Data quality check type to copy, the results of the other check types are ignored.
-     * @param includeColumns Include columns. When this parameter is false, the dictionary of columns is removed.
+     * @param checkFilter Check filter that filters the checks that should be preserved.
+     * @param includeColumnsAndChecks Include columns and checks. When this parameter is false, the dictionary of columns is removed.
      * @return A deep clone of the object with results only for that check type.
      */
-    public TableCurrentDataQualityStatusModel cloneFilteredByCheckType(CheckType checkType, boolean includeColumns) {
+    public TableCurrentDataQualityStatusModel cloneFilteredByCheckType(
+            Predicate<CheckCurrentDataQualityStatusModel> checkFilter, boolean includeColumnsAndChecks) {
         TableCurrentDataQualityStatusModel tableStatusClone = this.clone();
         tableStatusClone.currentSeverity = null;
         tableStatusClone.highestHistoricalSeverity = null;
@@ -321,14 +324,14 @@ public TableCurrentDataQualityStatusModel cloneFilteredByCheckType(CheckType che
         tableStatusClone.checks = new LinkedHashMap<>();
 
         for (Map.Entry<String, CheckCurrentDataQualityStatusModel> keyValue : this.checks.entrySet()) {
-            if (keyValue.getValue().getCheckType() == checkType) {
+            if (checkFilter == null || checkFilter.test(keyValue.getValue())) {
                 tableStatusClone.checks.put(keyValue.getKey(), keyValue.getValue());
             }
         }
 
         tableStatusClone.columns = new LinkedHashMap<>();
         for (Map.Entry<String, ColumnCurrentDataQualityStatusModel> columnKeyValue : this.columns.entrySet()) {
-            ColumnCurrentDataQualityStatusModel columnModelFiltered = columnKeyValue.getValue().cloneFilteredByCheckType(checkType);
+            ColumnCurrentDataQualityStatusModel columnModelFiltered = columnKeyValue.getValue().cloneFilteredByCheckType(checkFilter);
             if (!columnModelFiltered.getChecks().isEmpty()) {
                 tableStatusClone.columns.put(columnKeyValue.getKey(), columnModelFiltered);
             }
@@ -339,7 +342,8 @@ public TableCurrentDataQualityStatusModel cloneFilteredByCheckType(CheckType che
         tableStatusClone.calculateDataQualityKpiScore();
         tableStatusClone.calculateStatusesForDataQualityDimensions();
 
-        if (!includeColumns) {
+        if (!includeColumnsAndChecks) {
+            tableStatusClone.checks = null; // detaching checks
             tableStatusClone.columns = null; // detaching columns
         }
 

dqops/src/main/java/com/dqops/data/checkresults/statuscache/CurrentTableStatusCacheEntry.java

@@ -27,6 +27,7 @@ public class CurrentTableStatusCacheEntry {
     private final Object lock = new Object();
     private CurrentTableStatusEntryStatus status;
     private TableCurrentDataQualityStatusModel allCheckTypesWithColumns;
+    private TableCurrentDataQualityStatusModel monitoringAndPartitionedTableOnly;
     private TableCurrentDataQualityStatusModel profilingTableOnly;
     private TableCurrentDataQualityStatusModel monitoringTableOnly;
     private TableCurrentDataQualityStatusModel partitionedTableOnly;
@@ -77,6 +78,14 @@ public TableCurrentDataQualityStatusModel getAllCheckTypesWithColumns() {
         return allCheckTypesWithColumns;
     }
 
+    /**
+     * Returns a table data quality status that covers only monitoring and partitioned checks, but excludes the profiling status.
+     * @return The quality status for monitoring and partitioned checks only.
+     */
+    public TableCurrentDataQualityStatusModel getMonitoringAndPartitionedTableOnly() {
+        return monitoringAndPartitionedTableOnly;
+    }
+
     /**
      * Returns the status for the table related only to profiling checks. The status does not contain a list of column.
      * @return The status of the table, for profiling checks only.
@@ -103,15 +112,21 @@ public TableCurrentDataQualityStatusModel getPartitionedTableOnly() {
 
     /**
      * Sets the current model with a detailed status for all check types. If it is not empty, the status is also changed to loaded.
-     * @param allCheckTypesWithColumns Status model.
+     * @param allCheckTypesWithColumns Status model with all checks and columns.
+     * @param monitoringAndPartitionedTableOnly Status model with monitoring and partitioned checks, without columns.
+     * @param profilingTableOnly Status model with only profiling checks and no columns.
+     * @param monitoringTableOnly Status model with only monitoring checks and no columns.
+     * @param partitionedTableOnly Status model with only partitioned checks and no columns.
      */
     public void setStatusModels(
             TableCurrentDataQualityStatusModel allCheckTypesWithColumns,
+            TableCurrentDataQualityStatusModel monitoringAndPartitionedTableOnly,
             TableCurrentDataQualityStatusModel profilingTableOnly,
             TableCurrentDataQualityStatusModel monitoringTableOnly,
             TableCurrentDataQualityStatusModel partitionedTableOnly) {
         synchronized (this.lock) {
             this.allCheckTypesWithColumns = allCheckTypesWithColumns;
+            this.monitoringAndPartitionedTableOnly = monitoringAndPartitionedTableOnly;
             this.profilingTableOnly = profilingTableOnly;
             this.monitoringTableOnly = monitoringTableOnly;
             this.partitionedTableOnly = partitionedTableOnly;

dqops/src/main/java/com/dqops/data/checkresults/statuscache/TableStatusCache.java

@@ -31,11 +31,19 @@ public interface TableStatusCache {
      *
      * @param tableStatusKey Table status key.
      * @param checkType Check type. When a check type is given, the operation returns the status model only for one check type. The returned model does not contain columns.
-     *                  When the <code>checkType</code> is null, this method returns a full model for all checks, including all columns.
+     *                  When the <code>checkType</code> is null, this method returns a model for monitoring and partitioned checks combined.
      * @return Table status model or null when it is not yet loaded.
      */
     TableCurrentDataQualityStatusModel getCurrentTableStatus(CurrentTableStatusKey tableStatusKey, CheckType checkType);
 
+    /**
+     * Retrieves the current table status for a requested table including all columns.
+     *
+     * @param tableStatusKey Table status key.
+     * @return Table status model or null when it is not yet loaded.
+     */
+    TableCurrentDataQualityStatusModel getCurrentTableStatusWithColumns(CurrentTableStatusKey tableStatusKey);
+
     /**
      * Notifies the table status cache that the table result were updated and should be invalidated.
      *
@@ -47,9 +55,10 @@ public interface TableStatusCache {
 
     /**
      * Returns a future that is completed when there are no queued table status reload operations.
+     * @param waitTimeoutMilliseconds Optional timeout to wait for the completion of the future. If the timeout elapses, the future is completed with a value <code>false</code>.
      * @return Future that is completed when the status of all requested tables was loaded.
      */
-    CompletableFuture<Integer> getQueueEmptyFuture();
+    CompletableFuture<Boolean> getQueueEmptyFuture(Long waitTimeoutMilliseconds);
 
     /**
      * Starts a service that loads table statuses of requested tables.

dqops/src/main/java/com/dqops/data/checkresults/statuscache/TableStatusCacheImpl.java

@@ -101,18 +101,30 @@ protected CurrentTableStatusCacheEntry loadEntryCore(CurrentTableStatusKey table
         return currentTableStatusCacheEntry;
     }
 
+    /**
+     * Retrieves the current table status for a requested table including all columns.
+     *
+     * @param tableStatusKey Table status key.
+     * @return Table status model or null when it is not yet loaded.
+     */
+    @Override
+    public TableCurrentDataQualityStatusModel getCurrentTableStatusWithColumns(CurrentTableStatusKey tableStatusKey) {
+        CurrentTableStatusCacheEntry currentTableStatusCacheEntry = this.tableStatusCache.get(tableStatusKey, this::loadEntryCore);
+        return currentTableStatusCacheEntry.getAllCheckTypesWithColumns();
+    }
+
     /**
      * Retrieves the current table status for a requested table.
      * @param tableStatusKey Table status key.
      * @param checkType Check type. When a check type is given, the operation returns the status model only for one check type. The returned model does not contain columns.
-     *                  When the <code>checkType</code> is null, this method returns a full model for all checks, including all columns.
+     *                  When the <code>checkType</code> is null, this method returns a model for monitoring and partitioned checks combined.
      * @return Table status model or null when it is not yet loaded.
      */
     @Override
     public TableCurrentDataQualityStatusModel getCurrentTableStatus(CurrentTableStatusKey tableStatusKey, CheckType checkType) {
         CurrentTableStatusCacheEntry currentTableStatusCacheEntry = this.tableStatusCache.get(tableStatusKey, this::loadEntryCore);
         if (checkType == null) {
-            return currentTableStatusCacheEntry.getAllCheckTypesWithColumns();
+            return currentTableStatusCacheEntry.getMonitoringAndPartitionedTableOnly();
         }
 
         switch (checkType) {
@@ -153,12 +165,20 @@ public void invalidateTableStatus(CurrentTableStatusKey tableStatusKey, boolean
 
     /**
      * Returns a future that is completed when there are no queued table status reload operations.
+     * @param waitTimeoutMilliseconds Optional timeout to wait for the completion of the future. If the timeout elapses, the future is completed with a value <code>false</code>.
      * @return Future that is completed when the status of all requested tables was loaded.
      */
     @Override
-    public CompletableFuture<Integer> getQueueEmptyFuture() {
+    public CompletableFuture<Boolean> getQueueEmptyFuture(Long waitTimeoutMilliseconds) {
         synchronized (this.lock) {
-            return this.queueEmptyFuture;
+            CompletableFuture<Boolean> booleanCompletableFuture = this.queueEmptyFuture
+                    .thenApply(result -> true);
+
+            if (waitTimeoutMilliseconds != null) {
+                booleanCompletableFuture = booleanCompletableFuture.completeOnTimeout(false, waitTimeoutMilliseconds, TimeUnit.MILLISECONDS);
+            }
+
+            return booleanCompletableFuture;
         }
     }
 
@@ -210,11 +230,18 @@ public void onRequestLoadTableStatus(CurrentTableStatusKey tableStatusKey) {
 
             TableCurrentDataQualityStatusModel fullStatusModel =
                     this.checkResultsDataService.analyzeTableMostRecentQualityStatus(filterParameters, userDomainIdentity);
-            TableCurrentDataQualityStatusModel profilingStatus = fullStatusModel.cloneFilteredByCheckType(CheckType.profiling, false);
-            TableCurrentDataQualityStatusModel monitoringStatus = fullStatusModel.cloneFilteredByCheckType(CheckType.monitoring, false);
-            TableCurrentDataQualityStatusModel partitionedStatus = fullStatusModel.cloneFilteredByCheckType(CheckType.partitioned, false);
 
-            currentTableStatusCacheEntry.setStatusModels(fullStatusModel, profilingStatus, monitoringStatus, partitionedStatus); // also sets the status
+            TableCurrentDataQualityStatusModel monitoringAndPartitionedStatus = fullStatusModel.cloneFilteredByCheckType(
+                    checkResultsModel -> checkResultsModel.getCheckType() == CheckType.monitoring || checkResultsModel.getCheckType() == CheckType.partitioned, false);
+            TableCurrentDataQualityStatusModel profilingStatus = fullStatusModel.cloneFilteredByCheckType(
+                    checkResultsModel -> checkResultsModel.getCheckType() == CheckType.profiling, false);
+            TableCurrentDataQualityStatusModel monitoringStatus = fullStatusModel.cloneFilteredByCheckType(
+                    checkResultsModel -> checkResultsModel.getCheckType() == CheckType.monitoring, false);
+            TableCurrentDataQualityStatusModel partitionedStatus = fullStatusModel.cloneFilteredByCheckType(
+                    checkResultsModel -> checkResultsModel.getCheckType() == CheckType.partitioned, false);
+
+            currentTableStatusCacheEntry.setStatusModels(fullStatusModel, monitoringAndPartitionedStatus,
+                    profilingStatus, monitoringStatus, partitionedStatus); // also sets the status
         }
         catch (Exception ex) {
             currentTableStatusCacheEntry.setStatus(CurrentTableStatusEntryStatus.LOADED);

dqops/src/main/java/com/dqops/metadata/search/CheckSearchFiltersVisitor.java

@@ -358,6 +358,10 @@ public TreeNodeTraversalResult accept(AbstractCheckSpec<?,?,?,?> abstractCheckSp
 
         parameter.getNodes().add(abstractCheckSpec);
 
+        if (this.filters.getMaxResults() != null && parameter.getNodes().size() >= this.filters.getMaxResults()) {
+            return TreeNodeTraversalResult.STOP_TRAVERSAL;
+        }
+
         return TreeNodeTraversalResult.SKIP_CHILDREN; // no need to search any deeper, we have found what we were looking for
     }
 

dqops/src/main/java/com/dqops/metadata/search/ColumnSearchFilters.java

@@ -55,6 +55,9 @@ public class ColumnSearchFilters {
             "The labels are assigned on the labels screen and stored in the *labels* node in the *.dqotable.yaml* file.")
     private String[] labels;
 
+    @JsonPropertyDescription("Optional limit for the maximum number of results to return.")
+    private Integer maxResults;
+
     @JsonIgnore
     private SearchPattern connectionNameSearchPattern;
     @JsonIgnore
@@ -204,6 +207,22 @@ public void setLabels(String[] labels) {
         this.labels = labels;
     }
 
+    /**
+     * Sets the limit for the maximum number of results to return.
+     * @return Limit or null.
+     */
+    public Integer getMaxResults() {
+        return maxResults;
+    }
+
+    /**
+     * Sets the limit for the maximum number of results to return.
+     * @param maxResults New limit of rows or null to disable limiting the results.
+     */
+    public void setMaxResults(Integer maxResults) {
+        this.maxResults = maxResults;
+    }
+
     /**
      * Returns the {@link SearchPattern} related to <code>connectionName</code>.
      * Lazy getter, parses <code>connectionName</code> as a search pattern and returns parsed object.

dqops/src/main/java/com/dqops/metadata/search/ColumnSearchFiltersVisitor.java

@@ -275,12 +275,22 @@ public TreeNodeTraversalResult accept(ColumnSpec columnSpec, SearchParameterObje
         String columnNameFilter = this.filters.getColumnName();
         if (Strings.isNullOrEmpty(columnNameFilter)) {
             parameter.getNodes().add(columnSpec);
+
+            if (this.filters.getMaxResults() != null && parameter.getNodes().size() >= this.filters.getMaxResults()) {
+                return TreeNodeTraversalResult.STOP_TRAVERSAL;
+            }
+
             return TreeNodeTraversalResult.SKIP_CHILDREN;
         }
 
         String columnName = columnSpec.getHierarchyId().getLast().toString();
         if (StringPatternComparer.matchSearchPattern(columnName, columnNameFilter)) {
             parameter.getNodes().add(columnSpec);
+
+            if (this.filters.getMaxResults() != null && parameter.getNodes().size() >= this.filters.getMaxResults()) {
+                return TreeNodeTraversalResult.STOP_TRAVERSAL;
+            }
+
             return TreeNodeTraversalResult.SKIP_CHILDREN;
         }
 

dqops/src/main/java/com/dqops/metadata/search/StatisticsCollectorSearchFiltersVisitor.java

@@ -362,6 +362,10 @@ public TreeNodeTraversalResult accept(AbstractStatisticsCollectorSpec<?> abstrac
 
         parameter.getNodes().add(abstractStatisticsCollectorSpec);
 
+        if (this.filters.getMaxResults() != null && parameter.getNodes().size() >= this.filters.getMaxResults()) {
+            return TreeNodeTraversalResult.STOP_TRAVERSAL;
+        }
+
         return TreeNodeTraversalResult.SKIP_CHILDREN; // no need to search any deeper, we have found what we were looking for
     }