This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch
4258-use-data-lake-import-api-for-faster-cypress-test-data-setup-1
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/4258-use-data-lake-import-api-for-faster-cypress-test-data-setup-1
by this push:
new 18e4110207 refactor(#4258): Refactor Cypress datalake test setup to
use CSV import API
18e4110207 is described below
commit 18e4110207585f2b52152cb7e13192a56ab14e1d
Author: Philipp Zehnder <[email protected]>
AuthorDate: Fri Mar 13 22:32:51 2026 +0100
refactor(#4258): Refactor Cypress datalake test setup to use CSV import API
---
ui/cypress/support/utils/PrepareTestDataUtils.ts | 64 ++---
ui/cypress/support/utils/chart/ChartUtils.ts | 85 +++----
.../support/utils/chart/ChartWidgetTableUtils.ts | 15 +-
.../support/utils/dataset/DataLakeSeedUtils.ts | 275 +++++++++++++++++++++
ui/cypress/tests/chart/configuration.smoke.spec.ts | 2 -
.../chart/filterNumericalStringProperties.spec.ts | 32 +--
ui/cypress/tests/dataset/csvImport.spec.ts | 17 ++
.../src/lib/apis/datalake-rest.service.ts | 17 ++
.../src/lib/model/datalake/csv-import.model.ts | 11 +-
.../data-download-dialog.component.html | 2 +-
.../standard-dialog/standard-dialog.component.scss | 3 +-
.../csv-import-dialog.component.html | 126 ++++++----
.../csv-import-dialog.component.scss | 49 ++--
.../csv-import-dialog.component.ts | 190 ++++----------
14 files changed, 555 insertions(+), 333 deletions(-)
diff --git a/ui/cypress/support/utils/PrepareTestDataUtils.ts
b/ui/cypress/support/utils/PrepareTestDataUtils.ts
index be300a7932..da2bceaea2 100644
--- a/ui/cypress/support/utils/PrepareTestDataUtils.ts
+++ b/ui/cypress/support/utils/PrepareTestDataUtils.ts
@@ -16,10 +16,7 @@
*
*/
-import { FileManagementUtils } from './FileManagementUtils';
-import { ConnectUtils } from './connect/ConnectUtils';
-import { AdapterBuilder } from '../builder/AdapterBuilder';
-import { ConnectBtns } from './connect/ConnectBtns';
+import { DataLakeSeedUtils } from './dataset/DataLakeSeedUtils';
export class PrepareTestDataUtils {
public static dataName = 'prepared_data';
@@ -29,52 +26,25 @@ export class PrepareTestDataUtils {
format: 'csv' | 'json_array' = 'csv',
storeInDataLake: boolean = true,
) {
- // Create adapter with dataset
- FileManagementUtils.addFile(dataSet);
-
- const adapter = this.getDataLakeTestAdapter(
- PrepareTestDataUtils.dataName,
- format,
- storeInDataLake,
- );
-
- ConnectUtils.addAdapter(adapter);
-
- ConnectUtils.startAdapter(adapter, true);
- }
-
- private static getDataLakeTestAdapter(
- name: string,
- format: 'csv' | 'json_array',
- storeInDataLake: boolean = true,
- ) {
- const adapterBuilder = AdapterBuilder.create('File_Stream')
- .setName(name)
- .setTimestampProperty('timestamp')
- .addProtocolInput(
- 'radio',
- 'speed',
- 'fastest_\\(ignore_original_time\\)',
- )
- .addProtocolInput('radio', 'replayonce', 'yes');
+ if (!storeInDataLake) {
+ throw new Error(
+ 'Direct datalake test seeding only supports persisted
datasets.',
+ );
+ }
if (format === 'csv') {
- adapterBuilder
- .setFormat('csv')
- .addFormatInput('input', ConnectBtns.csvDelimiter(), ';')
- .addFormatInput('checkbox', ConnectBtns.csvHeader(), 'check');
+ return DataLakeSeedUtils.importCsvFixture({
+ fixture: dataSet,
+ measurementName: PrepareTestDataUtils.dataName,
+ delimiter: ';',
+ timestampColumn: 'timestamp',
+ });
} else {
- adapterBuilder
- .setFormat('json')
- .addFormatInput('radio', 'json_options-array', '');
- }
-
- adapterBuilder.setStartAdapter(true);
-
- if (storeInDataLake) {
- adapterBuilder.setStoreInDataLake();
+ return DataLakeSeedUtils.importJsonArrayFixture({
+ fixture: dataSet,
+ measurementName: PrepareTestDataUtils.dataName,
+ timestampColumn: 'timestamp',
+ });
}
-
- return adapterBuilder.build();
}
}
diff --git a/ui/cypress/support/utils/chart/ChartUtils.ts
b/ui/cypress/support/utils/chart/ChartUtils.ts
index 6bf716d333..8498b045be 100644
--- a/ui/cypress/support/utils/chart/ChartUtils.ts
+++ b/ui/cypress/support/utils/chart/ChartUtils.ts
@@ -20,13 +20,11 @@ import { DataLakeFilterConfig } from
'../../model/DataLakeFilterConfig';
import { ChartWidget } from '../../model/ChartWidget';
import { DataSetUtils } from '../DataSetUtils';
import { PrepareTestDataUtils } from '../PrepareTestDataUtils';
-import { FileManagementUtils } from '../FileManagementUtils';
-import { ConnectUtils } from '../connect/ConnectUtils';
-import { ConnectBtns } from '../connect/ConnectBtns';
-import { AdapterBuilder } from '../../builder/AdapterBuilder';
import { GeneralUtils } from '../GeneralUtils';
import { ChartBtns } from './ChartBtns';
import { SharedBtns } from '../shared/SharedBtns';
+import { DataLakeSeedUtils } from '../dataset/DataLakeSeedUtils';
+import { ConnectBtns } from '../connect/ConnectBtns';
export class ChartUtils {
public static ADAPTER_NAME = 'datalake_configuration';
@@ -86,52 +84,29 @@ export class ChartUtils {
ChartUtils.loadRandomDataSetIntoDataLake();
}
- public static getDataLakeTestSetAdapter(
- name: string,
- storeInDataLake: boolean = true,
- format: 'csv' | 'json_array',
- ) {
- const adapterBuilder = AdapterBuilder.create('File_Stream')
- .setName(name)
- .setTimestampProperty('timestamp')
- .addDimensionProperty('randomtext')
- .addProtocolInput(
- 'radio',
- 'speed',
- 'fastest_\\(ignore_original_time\\)',
- )
- .setStartAdapter(true);
-
- if (format === 'csv') {
- adapterBuilder
- .setFormat('csv')
- .addFormatInput('input', ConnectBtns.csvDelimiter(), ';')
- .addFormatInput('checkbox', ConnectBtns.csvHeader(), 'check');
- } else {
- adapterBuilder.setFormat('json_array');
- }
-
- if (storeInDataLake) {
- adapterBuilder.setStoreInDataLake();
- }
- return adapterBuilder.build();
- }
-
public static loadDataIntoDataLake(
dataSet: string,
format: 'csv' | 'json_array' = 'csv',
) {
- // Create adapter with dataset
- FileManagementUtils.addFile(dataSet);
-
- const adapter = this.getDataLakeTestSetAdapter(
- ChartUtils.ADAPTER_NAME,
- true,
- format,
- );
-
- ConnectUtils.addAdapter(adapter);
- ConnectUtils.startAdapter(adapter);
+ if (format === 'csv') {
+ return DataLakeSeedUtils.importCsvFixture({
+ fixture: dataSet,
+ measurementName: ChartUtils.ADAPTER_NAME,
+ delimiter: ';',
+ timestampColumn: 'timestamp',
+ columnOverrides: {
+ randomtext: {
+ propertyScope: 'DIMENSION_PROPERTY',
+ },
+ },
+ });
+ } else {
+ return DataLakeSeedUtils.importJsonArrayFixture({
+ fixture: dataSet,
+ measurementName: ChartUtils.ADAPTER_NAME,
+ timestampColumn: 'timestamp',
+ });
+ }
}
public static addDataViewAndWidget(
@@ -304,6 +279,8 @@ export class ChartUtils {
public static createAndEditDataView() {
// Create new data view
ChartBtns.openNewDataViewBtn().click();
+ cy.location('hash').should('include', '/chart/create');
+ cy.location('hash').should('include', 'editMode=true');
}
public static removeWidget(dataViewName: string) {
@@ -431,11 +408,17 @@ export class ChartUtils {
}
public static selectDataSet(dataSet: string) {
- cy.dataCy('data-explorer-select-data-set')
- .click()
- .get('mat-option')
- .contains(dataSet)
- .click();
+ cy.get('body').then($body => {
+ if (
+ $body.find('[data-cy="data-explorer-select-data-set"]').length
+ ) {
+ cy.dataCy('data-explorer-select-data-set')
+ .click()
+ .get('mat-option')
+ .contains(dataSet)
+ .click();
+ }
+ });
}
public static assertSelectDataSet(dataSet: string) {
diff --git a/ui/cypress/support/utils/chart/ChartWidgetTableUtils.ts
b/ui/cypress/support/utils/chart/ChartWidgetTableUtils.ts
index 1f71ce8d8f..96f2540001 100644
--- a/ui/cypress/support/utils/chart/ChartWidgetTableUtils.ts
+++ b/ui/cypress/support/utils/chart/ChartWidgetTableUtils.ts
@@ -17,10 +17,15 @@
*/
export class ChartWidgetTableUtils {
- public static chartTableRowTimestamp() {
- return cy.dataCy('data-explorer-table-row-timestamp', {
- timeout: 10000,
- });
+ public static chartTableRows() {
+ return cy
+ .dataCy('data-explorer-table', {
+ timeout: 10000,
+ })
+ .filter(':visible')
+ .first()
+ .find('tbody tr')
+ .not(':has(.table-empty-row)');
}
/**
@@ -28,6 +33,6 @@ export class ChartWidgetTableUtils {
* @param amount of expected rows
*/
public static checkAmountOfRows(amount: number) {
- this.chartTableRowTimestamp().should('have.length', amount);
+ this.chartTableRows().should('have.length', amount);
}
}
diff --git a/ui/cypress/support/utils/dataset/DataLakeSeedUtils.ts
b/ui/cypress/support/utils/dataset/DataLakeSeedUtils.ts
new file mode 100644
index 0000000000..a81fbd6c6f
--- /dev/null
+++ b/ui/cypress/support/utils/dataset/DataLakeSeedUtils.ts
@@ -0,0 +1,275 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+import * as CSV from 'csv-string';
+
+type CsvRuntimeType = 'STRING' | 'BOOLEAN' | 'LONG' | 'FLOAT';
+type CsvImportTargetMode = 'NEW' | 'EXISTING';
+
+interface CsvImportConfiguration {
+ delimiter: string;
+ decimalSeparator: '.' | ',';
+ hasHeader: boolean;
+}
+
+interface CsvImportColumn {
+ csvColumn: string;
+ runtimeName: string;
+ runtimeType: CsvRuntimeType;
+ propertyScope?: string;
+ semanticType?: string;
+ inferredType?: CsvRuntimeType;
+ timestampCandidate?: boolean;
+}
+
+interface CsvImportTarget {
+ mode: CsvImportTargetMode;
+ measurementName: string;
+}
+
+interface CsvImportPreviewResult {
+ headers: string[];
+ previewRows: string[][];
+ columns: CsvImportColumn[];
+}
+
+interface CsvImportResult {
+ importedRowCount: number;
+ validationMessages: Array<{ field: string; message: string }>;
+}
+
+interface ColumnOverride {
+ runtimeName?: string;
+ runtimeType?: CsvRuntimeType;
+ propertyScope?: string;
+ semanticType?: string;
+}
+
+interface CsvFixtureImportOptions {
+ fixture: string;
+ measurementName: string;
+ delimiter?: string;
+ decimalSeparator?: '.' | ',';
+ timestampColumn?: string;
+ columnOverrides?: Record<string, ColumnOverride>;
+}
+
+interface JsonArrayFixtureImportOptions {
+ fixture: string;
+ measurementName: string;
+ timestampColumn?: string;
+ columnOverrides?: Record<string, ColumnOverride>;
+}
+
+interface ImportRequest {
+ csvConfig: CsvImportConfiguration;
+ headers: string[];
+ rows: string[][];
+ target: CsvImportTarget;
+ timestampColumn: string;
+ columns: CsvImportColumn[];
+}
+
+export class DataLakeSeedUtils {
+ private static readonly TIMESTAMP_SEMANTIC_TYPE =
+ 'http://schema.org/DateTime';
+
+ public static importCsvFixture(
+ options: CsvFixtureImportOptions,
+ ): Cypress.Chainable<CsvImportResult> {
+ const delimiter = options.delimiter ?? ';';
+ const decimalSeparator = options.decimalSeparator ?? '.';
+
+ return cy.fixture(options.fixture, 'utf8').then((content: string) => {
+ const parseCsv = CSV.parse as any;
+ const parsedCsv = parseCsv(content, delimiter);
+ const headers = parsedCsv[0];
+ const rows = parsedCsv.slice(1);
+ const timestampColumn = options.timestampColumn ?? headers[0];
+
+ return this.previewAndImport({
+ headers,
+ rows,
+ csvConfig: {
+ delimiter,
+ decimalSeparator,
+ hasHeader: true,
+ },
+ measurementName: options.measurementName,
+ timestampColumn,
+ columnOverrides: options.columnOverrides,
+ });
+ });
+ }
+
+ public static importJsonArrayFixture(
+ options: JsonArrayFixtureImportOptions,
+ ): Cypress.Chainable<CsvImportResult> {
+ return cy.fixture(options.fixture).then((records: Array<any>) => {
+ const headers = this.extractHeaders(records);
+ const rows = records.map(record =>
+ headers.map(header =>
+ this.serializeCell(record ? record[header] : undefined),
+ ),
+ );
+ const timestampColumn = options.timestampColumn ?? headers[0];
+
+ return this.previewAndImport({
+ headers,
+ rows,
+ csvConfig: {
+ delimiter: ';',
+ decimalSeparator: '.',
+ hasHeader: true,
+ },
+ measurementName: options.measurementName,
+ timestampColumn,
+ columnOverrides: options.columnOverrides,
+ });
+ });
+ }
+
+ private static previewAndImport(options: {
+ headers: string[];
+ rows: string[][];
+ csvConfig: CsvImportConfiguration;
+ measurementName: string;
+ timestampColumn: string;
+ columnOverrides?: Record<string, ColumnOverride>;
+ }): Cypress.Chainable<CsvImportResult> {
+ const token = window.localStorage.getItem('auth-token');
+ const target = {
+ mode: 'NEW' as CsvImportTargetMode,
+ measurementName: options.measurementName,
+ };
+
+ return cy
+ .request<CsvImportPreviewResult>({
+ method: 'POST',
+ url: '/streampipes-backend/api/v4/datalake/import/preview',
+ body: {
+ csvConfig: options.csvConfig,
+ headers: options.headers,
+ rows: options.rows,
+ target,
+ },
+ headers: {
+ Authorization: `Bearer ${token}`,
+ },
+ })
+ .then(previewResponse => {
+ const columns = this.buildColumns(
+ previewResponse.body.columns,
+ options.timestampColumn,
+ options.columnOverrides ?? {},
+ );
+
+ const request: ImportRequest = {
+ csvConfig: options.csvConfig,
+ headers: options.headers,
+ rows: options.rows,
+ target,
+ timestampColumn: options.timestampColumn,
+ columns,
+ };
+
+ return cy
+ .request<CsvImportResult>({
+ method: 'POST',
+ url: '/streampipes-backend/api/v4/datalake/import',
+ body: request,
+ headers: {
+ Authorization: `Bearer ${token}`,
+ },
+ })
+ .then(importResponse => {
+ expect(
+ importResponse.body.validationMessages,
+ 'import validation messages',
+ ).to.have.length(0);
+ expect(
+ importResponse.body.importedRowCount,
+ 'imported row count',
+ ).to.equal(options.rows.length);
+ return importResponse.body;
+ });
+ });
+ }
+
+ private static buildColumns(
+ previewColumns: CsvImportColumn[],
+ timestampColumn: string,
+ columnOverrides: Record<string, ColumnOverride>,
+ ) {
+ return previewColumns.map(previewColumn => {
+ const override = columnOverrides[previewColumn.csvColumn] ?? {};
+ if (previewColumn.csvColumn === timestampColumn) {
+ return {
+ ...previewColumn,
+ runtimeName:
+ override.runtimeName ?? previewColumn.runtimeName,
+ runtimeType: override.runtimeType ?? 'LONG',
+ propertyScope: 'HEADER_PROPERTY',
+ semanticType:
+ override.semanticType ??
+ DataLakeSeedUtils.TIMESTAMP_SEMANTIC_TYPE,
+ };
+ }
+
+ return {
+ ...previewColumn,
+ runtimeName: override.runtimeName ?? previewColumn.runtimeName,
+ runtimeType:
+ override.runtimeType ??
+ this.defaultRuntimeType(previewColumn.inferredType),
+ propertyScope: override.propertyScope ??
'MEASUREMENT_PROPERTY',
+ semanticType: override.semanticType ?? undefined,
+ };
+ });
+ }
+
+ private static defaultRuntimeType(
+ inferredType: CsvRuntimeType = 'STRING',
+ ): CsvRuntimeType {
+ if (inferredType === 'LONG' || inferredType === 'FLOAT') {
+ return 'FLOAT';
+ }
+
+ return inferredType;
+ }
+
+ private static extractHeaders(records: Array<any>) {
+ const headers: string[] = [];
+ records.forEach(record => {
+ Object.keys(record || {}).forEach(key => {
+ if (headers.indexOf(key) === -1) {
+ headers.push(key);
+ }
+ });
+ });
+ return headers;
+ }
+
+ private static serializeCell(value: any): string {
+ if (value === undefined || value === null) {
+ return '';
+ }
+
+ return String(value);
+ }
+}
diff --git a/ui/cypress/tests/chart/configuration.smoke.spec.ts
b/ui/cypress/tests/chart/configuration.smoke.spec.ts
index 9080a80515..a09a831321 100644
--- a/ui/cypress/tests/chart/configuration.smoke.spec.ts
+++ b/ui/cypress/tests/chart/configuration.smoke.spec.ts
@@ -16,7 +16,6 @@
*
*/
-import { PipelineUtils } from '../../support/utils/pipeline/PipelineUtils';
import { ChartUtils } from '../../support/utils/chart/ChartUtils';
import { ChartBtns } from '../../support/utils/chart/ChartBtns';
import { GeneralUtils } from '../../support/utils/GeneralUtils';
@@ -56,7 +55,6 @@ describe('Delete data in datalake', () => {
before('Setup Test', () => {
cy.initStreamPipesTest();
ChartUtils.loadRandomDataSetIntoDataLake();
- PipelineUtils.deletePipeline('Persist prepared_data');
});
it('Perform Test', () => {
diff --git a/ui/cypress/tests/chart/filterNumericalStringProperties.spec.ts
b/ui/cypress/tests/chart/filterNumericalStringProperties.spec.ts
index 2bd393488b..f09f65a856 100644
--- a/ui/cypress/tests/chart/filterNumericalStringProperties.spec.ts
+++ b/ui/cypress/tests/chart/filterNumericalStringProperties.spec.ts
@@ -19,30 +19,24 @@
import { ChartUtils } from '../../support/utils/chart/ChartUtils';
import { ChartWidgetTableUtils } from
'../../support/utils/chart/ChartWidgetTableUtils';
import { DataLakeFilterConfig } from
'../../support/model/DataLakeFilterConfig';
-import { AdapterBuilder } from '../../support/builder/AdapterBuilder';
-import { ConnectBtns } from '../../support/utils/connect/ConnectBtns';
-import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
-import { FileManagementUtils } from '../../support/utils/FileManagementUtils';
import { ChartWidget } from '../../support/model/ChartWidget';
+import { DataLakeSeedUtils } from
'../../support/utils/dataset/DataLakeSeedUtils';
describe('Validate that filter works for numerical dimension property', () => {
beforeEach('Setup Test', () => {
cy.initStreamPipesTest();
-
- FileManagementUtils.addFile(
- 'datalake/filterNumericalStringProperties.csv',
- );
- const adapterInput = AdapterBuilder.create('File_Stream')
- .setName('Test Adapter')
- .setTimestampProperty('timestamp')
- .addDataTypeChange('dimensionKey', 'Integer')
- .addDimensionProperty('dimensionKey')
- .setStoreInDataLake()
- .setFormat('csv')
- .addFormatInput('input', ConnectBtns.csvDelimiter(), ';')
- .addFormatInput('checkbox', ConnectBtns.csvHeader(), 'check')
- .build();
- ConnectUtils.testAdapter(adapterInput);
+ DataLakeSeedUtils.importCsvFixture({
+ fixture: 'datalake/filterNumericalStringProperties.csv',
+ measurementName: 'Test Adapter',
+ delimiter: ';',
+ timestampColumn: 'timestamp',
+ columnOverrides: {
+ dimensionKey: {
+ runtimeType: 'LONG',
+ propertyScope: 'DIMENSION_PROPERTY',
+ },
+ },
+ });
});
it('Perform Test', () => {
diff --git a/ui/cypress/tests/dataset/csvImport.spec.ts
b/ui/cypress/tests/dataset/csvImport.spec.ts
index 63f7a9f9c9..f7ff816887 100644
--- a/ui/cypress/tests/dataset/csvImport.spec.ts
+++ b/ui/cypress/tests/dataset/csvImport.spec.ts
@@ -22,6 +22,7 @@ describe('CSV import happy path', () => {
const datasetName = 'csv_machine_data_import';
const stringTimestampDatasetName = 'csv_machine_data_import_string_ts';
const existingDatasetName = 'csv_machine_data_existing_import';
+ const missingValuesDatasetName = 'csv_machine_data_missing_values';
beforeEach('Setup Test', () => {
cy.initStreamPipesTest();
@@ -57,6 +58,22 @@ describe('CSV import happy path', () => {
);
});
+ it('Uploads a CSV file with missing values and still imports all rows', ()
=> {
+ DatasetUtils.openCsvImportDialog();
+ DatasetUtils.uploadCsvImportFile(
+ 'datalake/machine-data-simulator-import-missing-values.csv',
+ );
+ DatasetUtils.createNewDatasetFromCsv(missingValuesDatasetName);
+ DatasetUtils.continueCsvImportToPreview();
+ DatasetUtils.selectCsvImportDelimiterComma();
+ DatasetUtils.selectCsvImportTimestampColumn(0);
+ DatasetUtils.uploadCsvImport();
+ DatasetUtils.expectDatasetTotalEventCount(
+ missingValuesDatasetName,
+ '7',
+ );
+ });
+
it('Appends matching data to an existing dataset and warns on mismatched
timestamp schema', () => {
DatasetUtils.openCsvImportDialog();
DatasetUtils.uploadCsvImportFile(
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 cd92d0362c..4413701820 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
@@ -247,7 +247,24 @@ export class DatalakeRestService {
previewImport(
request: CsvImportPreviewRequest,
+ file?: File,
): Observable<CsvImportPreviewResult> {
+ if (file) {
+ const formData = new FormData();
+ formData.append('file', file, file.name);
+ formData.append(
+ 'request',
+ new Blob([JSON.stringify(request)], {
+ type: 'application/json',
+ }),
+ );
+
+ return this.http.post<CsvImportPreviewResult>(
+ `${this.dataLakeImportUrl}/preview`,
+ formData,
+ );
+ }
+
return this.http.post<CsvImportPreviewResult>(
`${this.dataLakeImportUrl}/preview`,
request,
diff --git
a/ui/projects/streampipes/platform-services/src/lib/model/datalake/csv-import.model.ts
b/ui/projects/streampipes/platform-services/src/lib/model/datalake/csv-import.model.ts
index 00f058fca2..98edf75c48 100644
---
a/ui/projects/streampipes/platform-services/src/lib/model/datalake/csv-import.model.ts
+++
b/ui/projects/streampipes/platform-services/src/lib/model/datalake/csv-import.model.ts
@@ -51,14 +51,16 @@ export interface CsvImportValidationMessage {
}
export interface CsvImportPreviewRequest {
+ uploadId?: string;
fileName?: string;
csvConfig: CsvImportConfiguration;
- headers: string[];
- rows: string[][];
+ headers?: string[];
+ rows?: string[][];
target?: CsvImportTarget;
}
export interface CsvImportPreviewResult {
+ uploadId?: string;
headers: string[];
previewRows: string[][];
columns: CsvImportColumn[];
@@ -94,9 +96,10 @@ export interface CsvImportSchemaValidationResult {
}
export interface CsvImportRequest {
+ uploadId?: string;
csvConfig: CsvImportConfiguration;
- headers: string[];
- rows: string[][];
+ headers?: string[];
+ rows?: string[][];
target: CsvImportTarget;
timestampColumn: string;
columns: CsvImportColumn[];
diff --git
a/ui/projects/streampipes/shared-ui/src/lib/dialog/data-download-dialog/data-download-dialog.component.html
b/ui/projects/streampipes/shared-ui/src/lib/dialog/data-download-dialog/data-download-dialog.component.html
index 723a79fbc3..ebd0780566 100644
---
a/ui/projects/streampipes/shared-ui/src/lib/dialog/data-download-dialog/data-download-dialog.component.html
+++
b/ui/projects/streampipes/shared-ui/src/lib/dialog/data-download-dialog/data-download-dialog.component.html
@@ -53,7 +53,7 @@
</div>
<mat-divider></mat-divider>
- <div class="sp-dialog-actions actions-align-right">
+ <div class="sp-dialog-actions actions-align-left">
<button
mat-button
mat-flat-button
diff --git
a/ui/projects/streampipes/shared-ui/src/lib/dialog/standard-dialog/standard-dialog.component.scss
b/ui/projects/streampipes/shared-ui/src/lib/dialog/standard-dialog/standard-dialog.component.scss
index 4af62bebaf..9ce08157de 100644
---
a/ui/projects/streampipes/shared-ui/src/lib/dialog/standard-dialog/standard-dialog.component.scss
+++
b/ui/projects/streampipes/shared-ui/src/lib/dialog/standard-dialog/standard-dialog.component.scss
@@ -47,7 +47,8 @@ standard-dialog-container {
.dialog-panel-content {
height: 100%;
- overflow-y: auto;
+ overflow: auto;
+ min-width: 0;
}
.dialog-title {
diff --git
a/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.html
b/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.html
index 92b33e8de6..3f2bd24950 100644
---
a/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.html
+++
b/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.html
@@ -227,12 +227,6 @@
[level]="3"
[title]="'Live preview' | translate"
>
- @if (currentTarget) {
- <div section-actions class="target-pill">
- {{ currentTargetLabel }}
- </div>
- }
-
@if (hasSchemaMismatch) {
<sp-alert-banner
type="error"
@@ -261,6 +255,18 @@
}
}
+ @if (showTimestampSelectionWarning) {
+ <sp-alert-banner
+ type="warning"
+ data-cy="csv-import-timestamp-warning"
+ [title]="'Warning' | translate"
+ [description]="
+ 'Please select exactly one timestamp column
before uploading the CSV.'
+ | translate
+ "
+ ></sp-alert-banner>
+ }
+
@if (hasPreview) {
<div
class="preview-table-wrapper"
@@ -300,50 +306,46 @@
data-cy="csv-import-timestamp-format"
/>
</mat-form-field>
- }
- <mat-form-field
- appearance="outline"
- class="w-100"
- >
- <mat-select
- [value]="
- model.column
-
.runtimeType ||
- model.column
-
.inferredType ||
- 'STRING'
- "
- [disabled]="
-
isTimestampColumn(
- model
- )
- "
- (selectionChange)="
- setColumnType(
- model,
-
$event.value
- )
- "
-
data-cy="csv-import-column-type"
+ } @else {
+ <mat-form-field
+
appearance="outline"
+ class="w-100"
>
- <mat-option
- value="STRING"
-
>STRING</mat-option
- >
- <mat-option
- value="BOOLEAN"
-
>BOOLEAN</mat-option
- >
- <mat-option
- value="LONG"
-
>LONG</mat-option
- >
- <mat-option
- value="FLOAT"
-
>FLOAT</mat-option
+ <mat-select
+ [value]="
+
model.column
+
.runtimeType ||
+
model.column
+
.inferredType ||
+ 'STRING'
+ "
+
(selectionChange)="
+
setColumnType(
+ model,
+
$event.value
+ )
+ "
+
data-cy="csv-import-column-type"
>
- </mat-select>
- </mat-form-field>
+ <mat-option
+
value="STRING"
+
>STRING</mat-option
+ >
+ <mat-option
+
value="BOOLEAN"
+
>BOOLEAN</mat-option
+ >
+ <mat-option
+
value="LONG"
+
>LONG</mat-option
+ >
+ <mat-option
+
value="FLOAT"
+
>FLOAT</mat-option
+ >
+ </mat-select>
+ </mat-form-field>
+ }
<mat-form-field
appearance="outline"
class="w-100"
@@ -458,6 +460,32 @@
{{ importResult?.importedRowCount }}
{{ 'rows imported' | translate }}
</div>
+ } @else if (hasUploadError) {
+ <sp-alert-banner
+ type="error"
+ data-cy="csv-import-upload-error"
+ [title]="'Import failed' | translate"
+ [description]="uploadErrorMessages[0].message"
+ ></sp-alert-banner>
+
+ @if (uploadErrorMessages.length > 1) {
+ <ul
+ class="schema-mismatch-list"
+ data-cy="csv-import-upload-error-list"
+ >
+ @for (
+ message of
uploadErrorMessages.slice(1);
+ track message.message
+ ) {
+ <li
+ class="schema-mismatch-item"
+
data-cy="csv-import-upload-error-item"
+ >
+ {{ message.message }}
+ </li>
+ }
+ </ul>
+ }
}
</div>
</sp-split-section>
@@ -471,7 +499,7 @@
<mat-divider></mat-divider>
- <div class="sp-dialog-actions actions-align-right">
+ <div class="sp-dialog-actions actions-align-left">
<button
mat-button
mat-flat-button
diff --git
a/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.scss
b/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.scss
index c598628067..a40c8a13d3 100644
---
a/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.scss
+++
b/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.scss
@@ -10,6 +10,27 @@
overflow: auto;
}
+:host {
+ display: block;
+ width: 100%;
+ min-width: 0;
+}
+
+:host ::ng-deep sp-split-section {
+ display: block;
+ width: 100%;
+ min-width: 0;
+ max-width: 100%;
+}
+
+:host ::ng-deep sp-split-section .section-outer,
+:host ::ng-deep sp-split-section .section-body,
+:host ::ng-deep sp-split-section .section-body-compact {
+ width: 100%;
+ min-width: 0;
+ max-width: 100%;
+}
+
.section-disabled {
opacity: 0.65;
}
@@ -86,15 +107,23 @@
}
.preview-table-wrapper {
- overflow: auto;
+ display: block;
+ width: 100%;
+ max-width: 100%;
+ overflow-x: auto;
+ overflow-y: auto;
border: 1px solid var(--color-border, #d8dee4);
border-radius: 14px;
+ padding-bottom: 4px;
+ -webkit-overflow-scrolling: touch;
+ box-sizing: border-box;
}
.preview-table {
- width: 100%;
+ width: max-content;
+ min-width: 100%;
border-collapse: collapse;
- min-width: 900px;
+ table-layout: auto;
}
.preview-table th,
@@ -118,7 +147,8 @@
display: flex;
flex-direction: column;
gap: 10px;
- min-width: 220px;
+ min-width: 240px;
+ width: 240px;
}
.header-title {
@@ -136,17 +166,6 @@
font-size: 12px;
}
-.target-pill {
- display: inline-flex;
- align-items: center;
- padding: 4px 10px;
- border-radius: 999px;
- background: #eef4ff;
- color: #1849a9;
- font-size: 12px;
- font-weight: 600;
-}
-
.message-box {
border-radius: 12px;
background: #fff4e5;
diff --git
a/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.ts
b/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.ts
index aa98238d6c..3878443b76 100644
--- a/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.ts
+++ b/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.ts
@@ -102,11 +102,10 @@ export class CsvImportDialogComponent {
private previewReloadTimeout?: ReturnType<typeof setTimeout>;
private schemaValidationTimeout?: ReturnType<typeof setTimeout>;
+ selectedFile?: File;
+ uploadId?: string;
fileName = '';
- rawFileContent = '';
timestampFormat = '';
- parsedHeaders: string[] = [];
- parsedRows: string[][] = [];
previewResult?: CsvImportPreviewResult;
schemaValidationResult?: CsvImportSchemaValidationResult;
importResult?: CsvImportResult;
@@ -116,8 +115,8 @@ export class CsvImportDialogComponent {
localMessages: CsvImportValidationMessage[] = [];
parseForm = this.fb.group({
- delimiter: [';', Validators.required],
- decimalSeparator: [',' as ',' | '.', Validators.required],
+ delimiter: [',' as ',' | ';' | '|' | '\\t', Validators.required],
+ decimalSeparator: ['.' as ',' | '.', Validators.required],
hasHeader: [true, Validators.required],
});
@@ -180,8 +179,16 @@ export class CsvImportDialogComponent {
return !!this.importResult?.measurementName;
}
- get previewColumns(): string[] {
- return this.previewResult?.headers ?? [];
+ get uploadErrorMessages(): CsvImportValidationMessage[] {
+ if (this.importLoading || this.hasImportResult) {
+ return [];
+ }
+
+ return this.importResult?.validationMessages ?? [];
+ }
+
+ get hasUploadError(): boolean {
+ return this.uploadErrorMessages.length > 0;
}
get previewRows(): string[][] {
@@ -198,16 +205,8 @@ export class CsvImportDialogComponent {
)?.column.runtimeName;
}
- get canConfigureColumns(): boolean {
- return this.hasPreview;
- }
-
- get canConfigureParse(): boolean {
- return this.isTargetValid;
- }
-
get canProceedToConfiguration(): boolean {
- return this.isTargetValid && !!this.rawFileContent;
+ return this.isTargetValid && !!this.selectedFile;
}
get canImport(): boolean {
@@ -220,6 +219,10 @@ export class CsvImportDialogComponent {
);
}
+ get showTimestampSelectionWarning(): boolean {
+ return this.hasPreview && !this.selectedTimestampColumn;
+ }
+
get selectedTimestampColumnModel(): CsvImportColumnModel | undefined {
return this.columnModels.find(model => this.isTimestampColumn(model));
}
@@ -263,13 +266,6 @@ export class CsvImportDialogComponent {
);
}
- get currentTargetLabel(): string {
- if (!this.currentTarget) {
- return '-';
- }
- return this.currentTarget.measurementName;
- }
-
onFileSelected(event: Event): void {
const input = event.target as HTMLInputElement;
const file = input.files?.[0];
@@ -278,20 +274,17 @@ export class CsvImportDialogComponent {
}
this.fileName = file.name;
+ this.selectedFile = file;
+ this.uploadId = undefined;
this.timestampFormat = '';
- const reader = new FileReader();
- reader.onload = () => {
- this.rawFileContent = `${reader.result ?? ''}`;
- this.invalidatePreview();
- };
- reader.readAsText(file);
+ this.invalidatePreview();
}
nextStep(): void {
if (this.csvImportStepper.selectedIndex === 0) {
if (!this.canProceedToConfiguration) {
this.localMessages = this.validateLocalTarget();
- if (!this.rawFileContent) {
+ if (!this.selectedFile) {
this.localMessages.push({
field: 'file',
message: 'Please select a CSV file first.',
@@ -339,34 +332,24 @@ export class CsvImportDialogComponent {
return;
}
- if (!this.rawFileContent) {
+ if (!this.selectedFile && !this.uploadId) {
this.localMessages = [
{ field: 'file', message: 'Please select a CSV file first.' },
];
return;
}
- try {
- const { headers, rows } = this.parseCsv();
- this.parsedHeaders = headers;
- this.parsedRows = rows;
- } catch (error) {
- this.localMessages = [
- {
- field: 'file',
- message:
- 'The CSV file could not be parsed with the current
settings.',
- },
- ];
- return;
- }
-
this.previewLoading = true;
+ const useMultipartUpload = !!this.selectedFile && !this.uploadId;
this.datalakeRestService
- .previewImport(this.buildPreviewRequest(this.currentTarget))
+ .previewImport(
+ this.buildPreviewRequest(this.currentTarget),
+ useMultipartUpload ? this.selectedFile : undefined,
+ )
.subscribe({
next: preview => {
this.previewResult = preview;
+ this.uploadId = preview.uploadId ?? this.uploadId;
this.columnModels = preview.columns.map(column =>
this.toColumnModel(column),
);
@@ -469,9 +452,23 @@ export class CsvImportDialogComponent {
},
error: error => {
this.importLoading = false;
- this.importResult = error?.error as CsvImportResult;
- if (!this.hasImportResult) {
- this.csvImportStepper.previous();
+ this.importResult = (error?.error as CsvImportResult) ?? {
+ measurementId: '',
+ measurementName: '',
+ createdNewMeasurement: false,
+ importedRowCount: 0,
+ validationMessages: [],
+ };
+
+ if (!this.importResult.validationMessages?.length) {
+ this.importResult.validationMessages = [
+ {
+ field: 'upload',
+ message:
+ error?.error?.message ??
+ 'The CSV import failed. Please review the
import configuration and try again.',
+ },
+ ];
}
},
});
@@ -481,106 +478,21 @@ export class CsvImportDialogComponent {
this.dialogRef.close(refresh);
}
- private parseCsv(): { headers: string[]; rows: string[][] } {
- const delimiter = this.normalizeDelimiter(
- this.parseForm.get('delimiter')?.value ?? ';',
- );
- const parsed = this.parseCsvContent(this.rawFileContent, delimiter);
- const rows = parsed.filter(row =>
- row.some(cell => `${cell}`.trim() !== ''),
- );
- if (rows.length === 0) {
- throw new Error('CSV contains no rows');
- }
-
- const hasHeader = this.parseForm.get('hasHeader')?.value ?? true;
- let headers: string[];
- let contentRows: string[][];
-
- if (hasHeader) {
- headers = rows[0].map((header, index) =>
- this.normalizeHeader(
- index === 0 ? this.stripBom(header) : header,
- index,
- ),
- );
- contentRows = rows.slice(1);
- } else {
- headers = rows[0].map((_, index) => `column_${index + 1}`);
- contentRows = rows;
- }
-
- return { headers, rows: contentRows };
- }
-
- private normalizeDelimiter(value: string): string {
- return value === '\\t' ? '\t' : value;
- }
-
- private stripBom(value: string): string {
- return value.replace(/^\uFEFF/, '');
- }
-
- private normalizeHeader(value: string, index: number): string {
- const trimmed = value?.trim();
- return trimmed ? trimmed : `column_${index + 1}`;
- }
-
- private parseCsvContent(content: string, delimiter: string): string[][] {
- const rows: string[][] = [];
- let currentRow: string[] = [];
- let currentValue = '';
- let inQuotes = false;
-
- for (let i = 0; i < content.length; i++) {
- const char = content[i];
- const nextChar = content[i + 1];
-
- if (char === '"') {
- if (inQuotes && nextChar === '"') {
- currentValue += '"';
- i += 1;
- } else {
- inQuotes = !inQuotes;
- }
- } else if (!inQuotes && char === delimiter) {
- currentRow.push(currentValue);
- currentValue = '';
- } else if (!inQuotes && (char === '\n' || char === '\r')) {
- if (char === '\r' && nextChar === '\n') {
- i += 1;
- }
- currentRow.push(currentValue);
- rows.push(currentRow);
- currentRow = [];
- currentValue = '';
- } else {
- currentValue += char;
- }
- }
-
- currentRow.push(currentValue);
- rows.push(currentRow);
- return rows;
- }
-
private buildPreviewRequest(
target?: CsvImportTarget,
): CsvImportPreviewRequest {
return {
+ uploadId: this.uploadId,
fileName: this.fileName,
csvConfig: this.currentCsvConfig,
- headers: this.parsedHeaders,
- rows: this.parsedRows,
target,
};
}
private buildImportRequest(): CsvImportRequest {
return {
+ uploadId: this.uploadId,
csvConfig: this.currentCsvConfig,
- headers: this.parsedHeaders,
- rows: this.parsedRows,
target: this.currentTarget!,
timestampColumn: this.selectedTimestampColumn!,
columns: this.columnModels.map(model => model.column),
@@ -629,10 +541,10 @@ export class CsvImportDialogComponent {
private get currentCsvConfig(): CsvImportConfiguration {
return {
- delimiter: this.parseForm.get('delimiter')?.value ?? ';',
+ delimiter: this.parseForm.get('delimiter')?.value ?? ',',
decimalSeparator:
(this.parseForm.get('decimalSeparator')?.value as ',' | '.') ??
- ',',
+ '.',
hasHeader: this.parseForm.get('hasHeader')?.value ?? true,
timestampFormat: this.timestampFormat.trim() || undefined,
};
