This is an automated email from the ASF dual-hosted git repository.
SvenO3 pushed a commit to branch
4633-move-dataset-count-display-into-dataset-overview
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/4633-move-dataset-count-display-into-dataset-overview by this push:
new 3d6ac5a125 Add new endpoint to get all latest events
3d6ac5a125 is described below
commit 3d6ac5a1252c9f7a62c9e20136484e54d39474b8
Author: Sven Oehler <[email protected]>
AuthorDate: Thu Jun 25 11:33:07 2026 +0200
Add new endpoint to get all latest events
---
.../rest/impl/datalake/DataLakeResource.java | 52 +++++++++++++++++
.../src/lib/apis/datalake-rest.service.ts | 12 ++++
.../datalake-configuration.component.ts | 65 ++++------------------
3 files changed, 75 insertions(+), 54 deletions(-)
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeResource.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeResource.java
index aede5118d5..2f05b40687 100644
---
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeResource.java
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeResource.java
@@ -241,6 +241,39 @@ public class DataLakeResource extends
AbstractDataLakeResource {
return ok(results);
}
+ @PostMapping(
+ path = "/measurements/latest-events",
+ produces = MediaType.APPLICATION_JSON_VALUE,
+ consumes = MediaType.APPLICATION_JSON_VALUE)
+ @PreAuthorize("this.hasReadAuthority()")
+ @Operation(summary = "Get the latest event timestamp for measurement
series", tags = { "Data Lake" })
+ public ResponseEntity<?> getLatestEvents(@RequestBody List<String>
measurementNames) {
+ if (measurementNames == null) {
+ return badRequest();
+ }
+
+ var distinctMeasurementNames = measurementNames.stream()
+ .distinct()
+ .toList();
+
+ var unauthorizedMeasureName = distinctMeasurementNames.stream()
+ .filter(measureName -> !checkPermissionByName(measureName, "READ"))
+ .findFirst();
+ if (unauthorizedMeasureName.isPresent()) {
+ return badRequest(
+ String.format("No read permission for measurement %s",
unauthorizedMeasureName.get())
+ );
+ }
+
+ Map<String, Long> latestEvents = distinctMeasurementNames.stream()
+ .collect(Collectors.toMap(
+ measurementName -> measurementName,
+ this::getLatestEvent
+ ));
+
+ return ok(latestEvents);
+ }
+
@GetMapping(path = "/measurements/{measurementID}/download", produces =
MediaType.APPLICATION_OCTET_STREAM_VALUE)
@PreAuthorize("this.hasReadAuthority() and
this.checkPermissionByName(#measurementID, 'READ')")
@Operation(summary = "Download data from a single measurement series by a
given id", tags = {
@@ -393,6 +426,25 @@ public class DataLakeResource extends
AbstractDataLakeResource {
return new ProvidedRestQueryParams(measurementId, queryParamMap);
}
+ private Long getLatestEvent(String measurementName) {
+ Map<String, String> queryParams = Map.of(
+ QP_START_DATE, "0",
+ QP_END_DATE, String.valueOf(System.currentTimeMillis()),
+ QP_LIMIT, "1",
+ QP_ORDER, "DESC",
+ QP_MISSING_VALUE_BEHAVIOUR, "empty"
+ );
+
+ try {
+ return this.dataExplorerQueryManagement
+ .getData(new ProvidedRestQueryParams(measurementName, queryParams),
true)
+ .getLastTimestamp();
+ } catch (RuntimeException e) {
+ LOG.warn("Could not get latest event for measurement {}",
measurementName, e);
+ return 0L;
+ }
+ }
+
// Checks if the parameter for missing value behaviour is set
private boolean isIgnoreMissingValues(String missingValueBehaviour) {
return "ignore".equals(missingValueBehaviour);
diff --git
a/ui/projects/streampipes/platform-services/src/lib/apis/datalake-rest.service.ts
b/ui/projects/streampipes/platform-services/src/lib/apis/datalake-rest.service.ts
index 8ab764a7ea..ff7769a800 100644
---
a/ui/projects/streampipes/platform-services/src/lib/apis/datalake-rest.service.ts
+++
b/ui/projects/streampipes/platform-services/src/lib/apis/datalake-rest.service.ts
@@ -119,6 +119,18 @@ export class DatalakeRestService {
.pipe(map(response => response as SpQueryResult[]));
}
+ getLatestMeasurementEvents(
+ measurementNames: string[],
+ ): Observable<Record<string, number>> {
+ return this.http.post<Record<string, number>>(
+ `${this.dataLakeUrl}/measurements/latest-events`,
+ measurementNames,
+ {
+ context: new HttpContext().set(NGX_LOADING_BAR_IGNORED, true),
+ },
+ );
+ }
+
getData(
index: string,
queryParams: DatalakeQueryParameters,
diff --git
a/ui/src/app/dataset/components/datalake-configuration/datalake-configuration.component.ts
b/ui/src/app/dataset/components/datalake-configuration/datalake-configuration.component.ts
index c8977701f6..f6d042fd70 100644
---
a/ui/src/app/dataset/components/datalake-configuration/datalake-configuration.component.ts
+++
b/ui/src/app/dataset/components/datalake-configuration/datalake-configuration.component.ts
@@ -45,7 +45,6 @@ import {
DatasetSummaryDto,
ExportProviderService,
ExportProviderSettings,
- SpQueryResult,
} from '@streampipes/platform-services';
import { MatPaginator } from '@angular/material/paginator';
import { MatSort, MatSortHeader } from '@angular/material/sort';
@@ -90,7 +89,7 @@ import { MatProgressSpinner } from
'@angular/material/progress-spinner';
import { DatePipe, NgStyle } from '@angular/common';
import { StyleDirective } from '@ngbracket/ngx-layout/extended';
import { MatMenuItem } from '@angular/material/menu';
-import { catchError, forkJoin, map, of, Subscription } from 'rxjs';
+import { catchError, of, Subscription } from 'rxjs';
import { LastUpdatedFormatterService } from
'../../../core-services/time-formatting/last-updated-formatter.service';
@Component({
@@ -482,31 +481,17 @@ export class DatalakeConfigurationComponent
queryLastEventTimes(measurements: DataLakeConfigurationEntry[]): void {
this.applyLastEventLoadingStatus(measurements, true);
- forkJoin(
- measurements.map(measurement =>
- this.datalakeRestService
- .getData(
- measurement.name,
- {
- endDate: Date.now(),
- startDate: 0,
- limit: 1,
- order: 'DESC',
- missingValueBehaviour: 'empty',
- },
- true,
- )
- .pipe(
- map(result => this.extractLastTimestamp(result)),
- catchError(() => of(0)),
- ),
- ),
- ).subscribe(res => {
- this.applyLastEventLoadingStatus(measurements, false);
- measurements.forEach((measurement, index) => {
- measurement.lastEvent = res[index];
+ this.datalakeRestService
+ .getLatestMeasurementEvents(
+ measurements.map(measurement => measurement.name),
+ )
+ .pipe(catchError(() => of({} as Record<string, number>)))
+ .subscribe(latestEvents => {
+ this.applyLastEventLoadingStatus(measurements, false);
+ measurements.forEach(measurement => {
+ measurement.lastEvent = latestEvents[measurement.name] ??
0;
+ });
});
- });
}
applyLastEventLoadingStatus(
@@ -541,34 +526,6 @@ export class DatalakeConfigurationComponent
return entry;
}
- private extractLastTimestamp(result: SpQueryResult): number {
- if (result?.lastTimestamp) {
- return result.lastTimestamp;
- }
-
- const series = result?.allDataSeries?.find(
- dataSeries => dataSeries.rows?.length > 0,
- );
- const row = series?.rows?.[0];
- if (!row) {
- return 0;
- }
-
- const headers = result.headers?.length
- ? result.headers
- : series.headers;
- const timestampIndex = headers?.indexOf('time') ?? 0;
- return this.toTimestamp(row[timestampIndex >= 0 ? timestampIndex : 0]);
- }
-
- private toTimestamp(value: unknown): number {
- const timestamp =
- typeof value === 'number'
- ? value
- : new Date(String(value)).getTime();
- return Number.isNaN(timestamp) ? 0 : timestamp;
- }
-
private openRetentionLogDialog(measurement: DataLakeMeasure): void {
const dialogRef: DialogRef<DataRetentionLogDialogComponent> =
this.dialogService.open(DataRetentionLogDialogComponent, {
