Skip to content

Commit b85d782

Browse files
piotrczarnaspiotrczarnas
committed
A list of labels is maintained automatically for all loaded connection and table yaml files.
1 parent 2b5ad28 commit b85d782

21 files changed

+892
-34
lines changed

‎dqops/src/main/java/com/dqops/cli/ApplicationShutdownManagerImpl.java

+7-1
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import com.dqops.core.jobqueue.monitoring.DqoJobQueueMonitoringService;
2121
import com.dqops.core.scheduler.JobSchedulerService;
2222
import com.dqops.data.checkresults.statuscache.TableStatusCache;
23+
import com.dqops.metadata.labels.labelloader.LabelsIndexer;
2324
import lombok.extern.slf4j.Slf4j;
2425
import org.springframework.beans.factory.annotation.Autowired;
2526
import org.springframework.boot.SpringApplication;
@@ -38,6 +39,7 @@ public class ApplicationShutdownManagerImpl implements ApplicationShutdownManage
3839
private JobSchedulerService jobSchedulerService;
3940
private DqoJobQueueMonitoringService jobQueueMonitoringService;
4041
private TableStatusCache tableStatusCache;
42+
private LabelsIndexer labelsIndexer;
4143

4244
/**
4345
* Constructor that receives dependencies to services that should be notified to shut down.
@@ -47,20 +49,23 @@ public class ApplicationShutdownManagerImpl implements ApplicationShutdownManage
4749
* @param jobSchedulerService Job scheduler.
4850
* @param jobQueueMonitoringService Job queue monitoring service.
4951
* @param tableStatusCache Table status cache.
52+
* @param labelsIndexer Label indexer service that indexes all labels.
5053
*/
5154
@Autowired
5255
public ApplicationShutdownManagerImpl(ApplicationContext applicationContext,
5356
DqoJobQueue dqoJobQueue,
5457
ParentDqoJobQueue parentDqoJobQueue,
5558
JobSchedulerService jobSchedulerService,
5659
DqoJobQueueMonitoringService jobQueueMonitoringService,
57-
TableStatusCache tableStatusCache) {
60+
TableStatusCache tableStatusCache,
61+
LabelsIndexer labelsIndexer) {
5862
this.applicationContext = applicationContext;
5963
this.dqoJobQueue = dqoJobQueue;
6064
this.parentDqoJobQueue = parentDqoJobQueue;
6165
this.jobSchedulerService = jobSchedulerService;
6266
this.jobQueueMonitoringService = jobQueueMonitoringService;
6367
this.tableStatusCache = tableStatusCache;
68+
this.labelsIndexer = labelsIndexer;
6469
}
6570

6671
/**
@@ -72,6 +77,7 @@ public void initiateShutdown(int returnCode) {
7277
if (log.isDebugEnabled()) {
7378
log.debug("Shutdown initialized with a return code: " + returnCode);
7479
}
80+
this.labelsIndexer.stop();
7581
this.tableStatusCache.stop();
7682
this.dqoJobQueue.stop();
7783
this.parentDqoJobQueue.stop();

‎dqops/src/main/java/com/dqops/cli/CliInitializerImpl.java

+7-1
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import com.dqops.core.synchronization.status.FileSynchronizationChangeDetectionService;
3434
import com.dqops.data.checkresults.statuscache.TableStatusCache;
3535
import com.dqops.data.storage.TablesawParquetSupportFix;
36+
import com.dqops.metadata.labels.labelloader.LabelsIndexer;
3637
import com.dqops.metadata.storage.localfiles.userhome.LocalUserHomeCreator;
3738
import com.dqops.rest.server.LocalUrlAddresses;
3839
import com.dqops.services.timezone.DefaultTimeZoneProvider;
@@ -69,6 +70,7 @@ public class CliInitializerImpl implements CliInitializer {
6970
private RootConfigurationProperties rootConfigurationProperties;
7071
private DqoUserPrincipalProvider dqoUserPrincipalProvider;
7172
private TableStatusCache tableStatusCache;
73+
private LabelsIndexer labelsIndexer;
7274

7375
/**
7476
* Called by the dependency injection container to provide dependencies.
@@ -90,6 +92,7 @@ public class CliInitializerImpl implements CliInitializer {
9092
* @param rootConfigurationProperties Root configuration parameters that are mapped to parameters not configured without any prefix, such as --silent.
9193
* @param dqoUserPrincipalProvider User principal provider.
9294
* @param tableStatusCache Table status cache.
95+
* @param labelsIndexer Label indexer service that finds all labels.
9396
*/
9497
@Autowired
9598
public CliInitializerImpl(LocalUserHomeCreator localUserHomeCreator,
@@ -109,7 +112,8 @@ public CliInitializerImpl(LocalUserHomeCreator localUserHomeCreator,
109112
PythonVirtualEnvService pythonVirtualEnvService,
110113
RootConfigurationProperties rootConfigurationProperties,
111114
DqoUserPrincipalProvider dqoUserPrincipalProvider,
112-
TableStatusCache tableStatusCache) {
115+
TableStatusCache tableStatusCache,
116+
LabelsIndexer labelsIndexer) {
113117
this.localUserHomeCreator = localUserHomeCreator;
114118
this.dqoCloudApiKeyProvider = dqoCloudApiKeyProvider;
115119
this.terminalReader = terminalReader;
@@ -128,6 +132,7 @@ public CliInitializerImpl(LocalUserHomeCreator localUserHomeCreator,
128132
this.rootConfigurationProperties = rootConfigurationProperties;
129133
this.dqoUserPrincipalProvider = dqoUserPrincipalProvider;
130134
this.tableStatusCache = tableStatusCache;
135+
this.labelsIndexer = labelsIndexer;
131136
}
132137

133138
/**
@@ -209,6 +214,7 @@ public void initializeApp(String[] args) {
209214
}
210215
finally {
211216
this.jobQueueMonitoringService.start();
217+
this.labelsIndexer.start();
212218
this.tableStatusCache.start();
213219
this.dqoJobQueue.start();
214220
this.parentDqoJobQueue.start();

‎dqops/src/main/java/com/dqops/core/filesystem/cache/LocalFileSystemCacheImpl.java

+61-21
Original file line numberDiff line numberDiff line change
@@ -26,10 +26,14 @@
2626
import com.dqops.data.checkresults.statuscache.CurrentTableStatusKey;
2727
import com.dqops.data.checkresults.statuscache.TableStatusCache;
2828
import com.dqops.data.checkresults.statuscache.TableStatusCacheProvider;
29-
import com.dqops.data.storage.HivePartitionPathUtility;
3029
import com.dqops.data.storage.LoadedMonthlyPartition;
3130
import com.dqops.data.storage.ParquetPartitioningKeys;
31+
import com.dqops.metadata.labels.labelloader.LabelRefreshKey;
32+
import com.dqops.metadata.labels.labelloader.LabelRefreshTarget;
33+
import com.dqops.metadata.labels.labelloader.LabelsIndexer;
34+
import com.dqops.metadata.labels.labelloader.LabelsIndexerProvider;
3235
import com.dqops.metadata.sources.PhysicalTableName;
36+
import com.dqops.metadata.storage.localfiles.SpecFileNames;
3337
import com.dqops.utils.exceptions.DqoRuntimeException;
3438
import com.dqops.utils.reflection.ObjectMemorySizeUtility;
3539
import com.github.benmanes.caffeine.cache.Cache;
@@ -67,6 +71,7 @@ public class LocalFileSystemCacheImpl implements LocalFileSystemCache, Disposabl
6771
private final Object directoryWatchersLock = new Object();
6872
private final DqoCacheConfigurationProperties dqoCacheConfigurationProperties;
6973
private final TableStatusCacheProvider tableStatusCacheProvider;
74+
private final LabelsIndexerProvider labelsIndexerProvider;
7075
private final HomeLocationFindService homeLocationFindService;
7176
private final Path userHomeRootPath;
7277
private Instant nextFileChangeDetectionAt = Instant.now().minus(100L, ChronoUnit.MILLIS);
@@ -76,14 +81,17 @@ public class LocalFileSystemCacheImpl implements LocalFileSystemCache, Disposabl
7681
* Dependency injection constructor.
7782
* @param dqoCacheConfigurationProperties Cache configuration parameters.
7883
* @param tableStatusCacheProvider Table status cache provider.
84+
* @param labelsIndexerProvider Labels indexer, notified to update labels for loaded or modified connections and tables.
7985
* @param homeLocationFindService Service that finds the location of the user home, used to translate file paths of updated files to relative file paths in the user home.
8086
*/
8187
@Autowired
8288
public LocalFileSystemCacheImpl(DqoCacheConfigurationProperties dqoCacheConfigurationProperties,
8389
TableStatusCacheProvider tableStatusCacheProvider,
90+
LabelsIndexerProvider labelsIndexerProvider,
8491
HomeLocationFindService homeLocationFindService) {
8592
this.dqoCacheConfigurationProperties = dqoCacheConfigurationProperties;
8693
this.tableStatusCacheProvider = tableStatusCacheProvider;
94+
this.labelsIndexerProvider = labelsIndexerProvider;
8795
this.homeLocationFindService = homeLocationFindService;
8896
this.userHomeRootPath = homeLocationFindService.getUserHomePath() != null ? Path.of(homeLocationFindService.getUserHomePath()) : Path.of(".");
8997

@@ -280,10 +288,19 @@ public FileContent loadFileContent(Path filePath, Function<Path, FileContent> re
280288
if (this.dqoCacheConfigurationProperties.isEnable()) {
281289
this.startFolderWatcher(filePath.getParent());
282290
this.processFileChanges(false);
283-
return this.fileContentsCache.get(filePath, readFunction);
291+
FileContent fileContent = this.fileContentsCache.getIfPresent(filePath);
292+
if (fileContent != null) {
293+
return fileContent;
294+
}
295+
296+
fileContent = this.fileContentsCache.get(filePath, readFunction);
297+
this.invalidateTableStatusCacheAndLabelsIndexer(filePath, false);
298+
return fileContent;
284299
}
285300

286-
return readFunction.apply(filePath);
301+
FileContent fileContentWithoutCache = readFunction.apply(filePath);
302+
this.invalidateTableStatusCacheAndLabelsIndexer(filePath, false);
303+
return fileContentWithoutCache;
287304
}
288305

289306
/**
@@ -294,6 +311,7 @@ public FileContent loadFileContent(Path filePath, Function<Path, FileContent> re
294311
@Override
295312
public void storeFile(Path filePath, FileContent fileContent) {
296313
if (!this.dqoCacheConfigurationProperties.isEnable()) {
314+
this.invalidateTableStatusCacheAndLabelsIndexer(filePath, false);
297315
return;
298316
}
299317

@@ -303,7 +321,9 @@ public void storeFile(Path filePath, FileContent fileContent) {
303321
processFileChanges(true);
304322
}
305323

324+
boolean replacingCachedFile = this.fileContentsCache.getIfPresent(filePath) != null;
306325
this.fileContentsCache.put(filePath, fileContent);
326+
this.invalidateTableStatusCacheAndLabelsIndexer(filePath, replacingCachedFile);
307327

308328
Path parentFolderPath = filePath.getParent();
309329
this.fileListsCache.invalidate(parentFolderPath);
@@ -336,6 +356,7 @@ public LoadedMonthlyPartition getParquetFile(Path filePath) {
336356
@Override
337357
public void storeParquetFile(Path filePath, LoadedMonthlyPartition table) {
338358
if (!this.dqoCacheConfigurationProperties.isEnable()) {
359+
this.invalidateTableStatusCacheAndLabelsIndexer(filePath, false);
339360
return;
340361
}
341362

@@ -347,7 +368,7 @@ public void storeParquetFile(Path filePath, LoadedMonthlyPartition table) {
347368

348369
boolean replacingCachedFile = this.parquetFilesCache.getIfPresent(filePath) != null;
349370
this.parquetFilesCache.put(filePath, table);
350-
invalidateTableStatusCache(filePath, replacingCachedFile);
371+
this.invalidateTableStatusCacheAndLabelsIndexer(filePath, replacingCachedFile);
351372

352373
Path parentFolderPath = filePath.getParent();
353374
this.fileListsCache.invalidate(parentFolderPath);
@@ -369,7 +390,7 @@ public void removeFile(Path filePath) {
369390
this.fileContentsCache.invalidate(filePath);
370391
this.parquetFilesCache.invalidate(filePath);
371392
this.wasRecentlyInvalidated = true;
372-
invalidateTableStatusCache(filePath, true);
393+
this.invalidateTableStatusCacheAndLabelsIndexer(filePath, true);
373394

374395

375396
Path parentFolderPath = filePath.getParent();
@@ -392,7 +413,7 @@ public void removeFolder(Path folderPath) {
392413

393414
this.folderListsCache.invalidate(folderPath);
394415
this.fileListsCache.invalidate(folderPath);
395-
invalidateTableStatusCache(folderPath, true);
416+
this.invalidateTableStatusCacheAndLabelsIndexer(folderPath, true);
396417

397418
this.stopFolderWatcher(folderPath);
398419

@@ -414,7 +435,7 @@ public void invalidateFolder(Path folderPath) {
414435

415436
this.folderListsCache.invalidate(folderPath);
416437
this.fileListsCache.invalidate(folderPath);
417-
invalidateTableStatusCache(folderPath, true);
438+
this.invalidateTableStatusCacheAndLabelsIndexer(folderPath, true);
418439
}
419440

420441
/**
@@ -430,7 +451,7 @@ public void invalidateFile(Path filePath) {
430451
this.fileContentsCache.invalidate(filePath);
431452
this.parquetFilesCache.invalidate(filePath);
432453
this.wasRecentlyInvalidated = true;
433-
invalidateTableStatusCache(filePath, true);
454+
this.invalidateTableStatusCacheAndLabelsIndexer(filePath, true);
434455

435456
Path folderPath = filePath.getParent();
436457
if (folderPath != null) {
@@ -442,11 +463,12 @@ public void invalidateFile(Path filePath) {
442463
/**
443464
* Matches the path of an updated or deleted file to a parquet file for tables that are cached: check_results and errors.
444465
* If a table is detected whose data quality results were updated, triggers an invalidation of a current table cache.
466+
* Also detects when a connection or table yaml file were modified, and rescans their indexes.
445467
* @param filePath File path to a file that should be updated.
446468
* @param replacingCachedFile True when we are replacing a file that was already in a cache, false when a file is just placed into a cache,
447469
* and it is not a real invalidation, but just a notification that a file was just cached.
448470
*/
449-
public void invalidateTableStatusCache(Path filePath, boolean replacingCachedFile) {
471+
public void invalidateTableStatusCacheAndLabelsIndexer(Path filePath, boolean replacingCachedFile) {
450472
if (!filePath.startsWith(this.userHomeRootPath)) {
451473
return;
452474
}
@@ -462,21 +484,39 @@ public void invalidateTableStatusCache(Path filePath, boolean replacingCachedFil
462484
}
463485

464486
HomeFolderPath folder = homeFilePath.getFolder();
465-
if (folder.size() < 4 || !Objects.equals(BuiltInFolderNames.DATA, folder.get(0).getFileSystemName()) ||
466-
!(Objects.equals(BuiltInFolderNames.CHECK_RESULTS, folder.get(1).getFileSystemName()) ||
487+
if (folder.size() >= 4 && Objects.equals(BuiltInFolderNames.DATA, folder.get(0).getFileSystemName()) &&
488+
(Objects.equals(BuiltInFolderNames.CHECK_RESULTS, folder.get(1).getFileSystemName()) ||
467489
Objects.equals(BuiltInFolderNames.ERRORS, folder.get(1).getFileSystemName()))) {
468-
return; // not a parquet folder
469-
}
490+
// parquet file updated
470491

471-
String connectionNameFolder = folder.get(2).getFileSystemName();
472-
String schemaTableNameFolder = folder.get(3).getFileSystemName();
492+
String connectionNameFolder = folder.get(2).getFileSystemName();
493+
String schemaTableNameFolder = folder.get(3).getFileSystemName();
473494

474-
if (connectionNameFolder.startsWith(ParquetPartitioningKeys.CONNECTION + "=") && connectionNameFolder.length() > 2 &&
475-
schemaTableNameFolder.startsWith(ParquetPartitioningKeys.SCHEMA_TABLE + "=") && schemaTableNameFolder.length() > 2) {
476-
String decodedConnectionName = FileNameSanitizer.decodeFileSystemName(connectionNameFolder.substring(2));
477-
PhysicalTableName physicalTableName = PhysicalTableName.fromBaseFileName(schemaTableNameFolder.substring(2));
478-
TableStatusCache tableStatusCache = this.tableStatusCacheProvider.getTableStatusCache();
479-
tableStatusCache.invalidateTableStatus(new CurrentTableStatusKey(folder.getDataDomain(), decodedConnectionName, physicalTableName), replacingCachedFile);
495+
if (connectionNameFolder.startsWith(ParquetPartitioningKeys.CONNECTION + "=") && connectionNameFolder.length() > 2 &&
496+
schemaTableNameFolder.startsWith(ParquetPartitioningKeys.SCHEMA_TABLE + "=") && schemaTableNameFolder.length() > 2) {
497+
String decodedConnectionName = FileNameSanitizer.decodeFileSystemName(connectionNameFolder.substring(2));
498+
PhysicalTableName physicalTableName = PhysicalTableName.fromBaseFileName(schemaTableNameFolder.substring(2));
499+
TableStatusCache tableStatusCache = this.tableStatusCacheProvider.getTableStatusCache();
500+
tableStatusCache.invalidateTableStatus(new CurrentTableStatusKey(folder.getDataDomain(), decodedConnectionName, physicalTableName), replacingCachedFile);
501+
}
502+
}
503+
504+
if (folder.size() >= 2 && Objects.equals(BuiltInFolderNames.SOURCES, folder.get(0).getFileSystemName())) {
505+
String connectionName = folder.get(1).getObjectName();
506+
String fileName = homeFilePath.getFileName();
507+
LabelsIndexer labelsIndexer = this.labelsIndexerProvider.getLabelsIndexer();
508+
509+
if (Objects.equals(fileName, SpecFileNames.CONNECTION_SPEC_FILE_NAME_YAML)) {
510+
labelsIndexer.invalidateObject(
511+
new LabelRefreshKey(LabelRefreshTarget.CONNECTION, folder.getDataDomain(), connectionName, null),
512+
replacingCachedFile);
513+
} else if (fileName != null && fileName.endsWith(SpecFileNames.TABLE_SPEC_FILE_EXT_YAML)) {
514+
String bareFileName = fileName.substring(0, fileName.length() - SpecFileNames.TABLE_SPEC_FILE_EXT_YAML.length());
515+
PhysicalTableName physicalTableName = PhysicalTableName.fromBaseFileName(bareFileName);
516+
labelsIndexer.invalidateObject(
517+
new LabelRefreshKey(LabelRefreshTarget.TABLE, folder.getDataDomain(), connectionName, physicalTableName),
518+
replacingCachedFile);
519+
}
480520
}
481521
}
482522

‎dqops/src/main/java/com/dqops/data/checkresults/statuscache/CurrentTableStatusKey.java

+12-1
Original file line numberDiff line numberDiff line change
@@ -26,9 +26,20 @@
2626
@Data
2727
@lombok.EqualsAndHashCode
2828
@lombok.ToString
29-
public class CurrentTableStatusKey {
29+
public final class CurrentTableStatusKey {
30+
/**
31+
* Data domain name.
32+
*/
3033
private String dataDomain;
34+
35+
/**
36+
* Connection name.
37+
*/
3138
private String connectionName;
39+
40+
/**
41+
* Physical table name.
42+
*/
3243
private PhysicalTableName physicalTableName;
3344

3445
/**

‎dqops/src/main/java/com/dqops/metadata/labels/DataDomainLabelsContainer.java ‎dqops/src/main/java/com/dqops/metadata/labels/labelcontainers/DataDomainLabelsContainer.java

+1-4
Original file line numberDiff line numberDiff line change
@@ -14,12 +14,9 @@
1414
* limitations under the License.
1515
*/
1616

17-
package com.dqops.metadata.labels;
17+
package com.dqops.metadata.labels.labelcontainers;
1818

1919
import com.dqops.metadata.sources.PhysicalTableName;
20-
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
21-
import org.springframework.context.annotation.Scope;
22-
import org.springframework.stereotype.Component;
2320

2421
import java.util.HashMap;
2522
import java.util.Objects;

‎dqops/src/main/java/com/dqops/metadata/labels/GlobalLabelsContainer.java ‎dqops/src/main/java/com/dqops/metadata/labels/labelcontainers/GlobalLabelsContainer.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
* limitations under the License.
1515
*/
1616

17-
package com.dqops.metadata.labels;
17+
package com.dqops.metadata.labels.labelcontainers;
1818

1919
import com.dqops.metadata.sources.PhysicalTableName;
2020

‎dqops/src/main/java/com/dqops/metadata/labels/GlobalLabelsContainerImpl.java ‎dqops/src/main/java/com/dqops/metadata/labels/labelcontainers/GlobalLabelsContainerImpl.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
* limitations under the License.
1515
*/
1616

17-
package com.dqops.metadata.labels;
17+
package com.dqops.metadata.labels.labelcontainers;
1818

1919
import com.dqops.metadata.sources.PhysicalTableName;
2020
import org.springframework.beans.factory.config.ConfigurableBeanFactory;

0 commit comments

Comments
 (0)