You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ze...@apache.org on 2026/08/26 14:08:37 UTC
(streampipes) 02/02: fix(#4850): preserve CSV types for existing datasets
This is an automated email from the ASF dual-hosted git repository.
tenthe pushed a commit to branch 4850-fix-csv-imports-for-existing-double-columns
in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit d4e26c96399d822350e19ebd9b9435e5c42e1f67
Author: Philipp Zehnder <te...@users.noreply.github.com>
AuthorDate: Wed Aug 26 16:08:17 2026 +0200
fix(#4850): preserve CSV types for existing datasets
---
.../impl/datalake/importer/CsvImportParser.java | 18 ++++-
.../importer/CsvDataLakeImportServiceTest.java | 64 +++++++++++++++++
.../datalake/importer/CsvImportParserTest.java | 81 ++++++++++++++++++++++
.../support/utils/dataset/DatasetSeedUtils.ts | 8 ++-
.../src/lib/model/datalake/csv-import.model.ts | 8 ++-
.../csv-import-dialog.component.spec.ts | 55 +++++++++++++++
.../csv-import-dialog.component.ts | 42 ++++++++---
.../csv-import-preview-table.component.html | 6 ++
8 files changed, 266 insertions(+), 16 deletions(-)
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/importer/CsvImportParser.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/importer/CsvImportParser.java
index 772ff87691..98823dcc21 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/importer/CsvImportParser.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/importer/CsvImportParser.java
@@ -34,6 +34,7 @@ import org.apache.streampipes.vocabulary.XSD;
import java.io.IOException;
import java.io.PushbackReader;
+import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -379,9 +380,11 @@ class CsvImportParser {
}
return switch (finalRuntimeType(column, timestampColumn)) {
- case "BOOLEAN" -> Boolean.parseBoolean(trimmed.toLowerCase(Locale.ENGLISH));
- case "LONG" -> Long.parseLong(normalizeNumber(trimmed, config));
- case "FLOAT" -> Double.parseDouble(normalizeNumber(trimmed, config));
+ case "BOOLEAN" -> parseBoolean(trimmed);
+ case "INTEGER" -> new BigDecimal(normalizeNumber(trimmed, config)).intValueExact();
+ case "LONG" -> new BigDecimal(normalizeNumber(trimmed, config)).longValueExact();
+ case "FLOAT" -> Float.parseFloat(normalizeNumber(trimmed, config));
+ case "DOUBLE" -> Double.parseDouble(normalizeNumber(trimmed, config));
default -> trimmed;
};
} catch (RuntimeException e) {
@@ -415,6 +418,13 @@ class CsvImportParser {
return "true".equalsIgnoreCase(value.trim()) || "false".equalsIgnoreCase(value.trim());
}
+ private boolean parseBoolean(String value) {
+ if (!isBoolean(value)) {
+ throw new IllegalArgumentException("Value is not a boolean");
+ }
+ return Boolean.parseBoolean(value);
+ }
+
private boolean isNumber(String value, CsvImportConfiguration config) {
try {
Double.parseDouble(normalizeNumber(value, config));
@@ -477,8 +487,10 @@ class CsvImportParser {
private String toRuntimeTypeUri(String runtimeType) {
return switch (runtimeType) {
case "BOOLEAN" -> XSD.BOOLEAN.toString();
+ case "INTEGER" -> XSD.INTEGER.toString();
case "LONG" -> XSD.LONG.toString();
case "FLOAT" -> XSD.FLOAT.toString();
+ case "DOUBLE" -> XSD.DOUBLE.toString();
default -> XSD.STRING.toString();
};
}
diff --git a/streampipes-rest/src/test/java/org/apache/streampipes/rest/impl/datalake/importer/CsvDataLakeImportServiceTest.java b/streampipes-rest/src/test/java/org/apache/streampipes/rest/impl/datalake/importer/CsvDataLakeImportServiceTest.java
index 508f542dfd..2d44d97c69 100644
--- a/streampipes-rest/src/test/java/org/apache/streampipes/rest/impl/datalake/importer/CsvDataLakeImportServiceTest.java
+++ b/streampipes-rest/src/test/java/org/apache/streampipes/rest/impl/datalake/importer/CsvDataLakeImportServiceTest.java
@@ -107,6 +107,62 @@ class CsvDataLakeImportServiceTest {
.anyMatch(issue -> issue.getType() == CsvImportSchemaIssueType.COLUMN_TYPE_MISMATCH));
}
+ @Test
+ void shouldPreserveAllPrimitiveTypesForExistingMeasurement() {
+ var schemaManagement = mock(IDataExplorerSchemaManagement.class);
+ var dataWriter = mock(DataLakeDataWriter.class);
+ var service = new CsvDataLakeImportService(schemaManagement, dataWriter);
+
+ var existingMeasure = new DataLakeMeasure();
+ existingMeasure.setMeasureName("existing-measure");
+ existingMeasure.setTimestampField("s0::timestamp");
+ existingMeasure.setEventSchema(new EventSchema(List.of(
+ makePrimitive("integerValue", XSD.INTEGER.toString(), "MEASUREMENT_PROPERTY"),
+ makePrimitive("longValue", XSD.LONG.toString(), "MEASUREMENT_PROPERTY"),
+ makePrimitive("floatValue", XSD.FLOAT.toString(), "MEASUREMENT_PROPERTY"),
+ makePrimitive("doubleValue", XSD.DOUBLE.toString(), "MEASUREMENT_PROPERTY"),
+ makePrimitive("booleanValue", XSD.BOOLEAN.toString(), "DIMENSION_PROPERTY"),
+ makePrimitive("stringValue", XSD.STRING.toString(), "DIMENSION_PROPERTY")
+ )));
+ when(schemaManagement.getExistingMeasureByName("existing-measure"))
+ .thenReturn(Optional.of(existingMeasure));
+
+ var previewRequest = new CsvImportPreviewRequest();
+ previewRequest.setCsvConfig(makeCsvConfig());
+ previewRequest.setHeaders(List.of(
+ "timestamp",
+ "integerValue",
+ "longValue",
+ "floatValue",
+ "doubleValue",
+ "booleanValue",
+ "stringValue"
+ ));
+ previewRequest.setRows(List.of(List.of(
+ "1710000000000",
+ "17.0",
+ "2147483648.0",
+ "1.25",
+ "1.23456789012345",
+ "true",
+ "value"
+ )));
+ previewRequest.setTarget(makeTarget(CsvImportTargetMode.EXISTING, "existing-measure"));
+
+ var previewResult = service.preview(previewRequest);
+
+ assertTrue(previewResult.isValid());
+ assertEquals(List.of("LONG", "INTEGER", "LONG", "FLOAT", "DOUBLE", "BOOLEAN", "STRING"),
+ previewResult.getColumns().stream().map(CsvImportColumn::getRuntimeType).toList());
+
+ var schemaRequest = new CsvImportSchemaValidationRequest();
+ schemaRequest.setTarget(previewRequest.getTarget());
+ schemaRequest.setTimestampColumn("timestamp");
+ schemaRequest.setColumns(previewResult.getColumns());
+
+ assertTrue(service.validateSchema(schemaRequest).isValid());
+ }
+
@Test
void shouldCreateNewMeasurementAndReuseSharedWriter() {
var schemaManagement = mock(IDataExplorerSchemaManagement.class);
@@ -416,6 +472,14 @@ class CsvDataLakeImportServiceTest {
return new EventSchema(List.of(timestamp, temperature));
}
+ private EventPropertyPrimitive makePrimitive(String runtimeName, String runtimeType, String propertyScope) {
+ var property = new EventPropertyPrimitive();
+ property.setRuntimeName(runtimeName);
+ property.setRuntimeType(runtimeType);
+ property.setPropertyScope(propertyScope);
+ return property;
+ }
+
private CsvImportJobStatus awaitTerminalStatus(
CsvDataLakeImportService service,
String jobId,
diff --git a/streampipes-rest/src/test/java/org/apache/streampipes/rest/impl/datalake/importer/CsvImportParserTest.java b/streampipes-rest/src/test/java/org/apache/streampipes/rest/impl/datalake/importer/CsvImportParserTest.java
index 709f50b6f8..34836b3cef 100644
--- a/streampipes-rest/src/test/java/org/apache/streampipes/rest/impl/datalake/importer/CsvImportParserTest.java
+++ b/streampipes-rest/src/test/java/org/apache/streampipes/rest/impl/datalake/importer/CsvImportParserTest.java
@@ -21,6 +21,8 @@ package org.apache.streampipes.rest.impl.datalake.importer;
import org.apache.streampipes.model.datalake.importer.CsvImportColumn;
import org.apache.streampipes.model.datalake.importer.CsvImportConfiguration;
import org.apache.streampipes.model.datalake.importer.CsvImportRequest;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.vocabulary.XSD;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -31,6 +33,7 @@ import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -103,6 +106,84 @@ class CsvImportParserTest {
assertTrue(exception.getValidationMessages().get(0).getMessage().contains("missing a value for timestamp"));
}
+ @Test
+ void shouldBuildSchemaAndConvertAllPrimitiveTypes() {
+ var columns = List.of(
+ makeColumn("timestamp", "timestamp", "LONG"),
+ makeColumn("integerValue", "integerValue", "INTEGER"),
+ makeColumn("longValue", "longValue", "LONG"),
+ makeColumn("floatValue", "floatValue", "FLOAT"),
+ makeColumn("doubleValue", "doubleValue", "DOUBLE"),
+ makeColumn("booleanValue", "booleanValue", "BOOLEAN"),
+ makeColumn("stringValue", "stringValue", "STRING")
+ );
+ var request = new CsvImportRequest();
+ request.setCsvConfig(makeConfig(",", ".", true));
+ request.setTimestampColumn("timestamp");
+ request.setColumns(columns);
+ request.setRows(List.of(List.of(
+ "1710000000000",
+ "17.0",
+ "2147483648.0",
+ "1.25",
+ "1.23456789012345",
+ "true",
+ "value"
+ )));
+
+ var row = parser.toImportRows(request).get(0);
+
+ assertInstanceOf(Long.class, row.get(0));
+ assertEquals(17, row.get(1));
+ assertInstanceOf(Integer.class, row.get(1));
+ assertEquals(2147483648L, row.get(2));
+ assertInstanceOf(Long.class, row.get(2));
+ assertEquals(1.25F, row.get(3));
+ assertInstanceOf(Float.class, row.get(3));
+ assertEquals(1.23456789012345D, row.get(4));
+ assertInstanceOf(Double.class, row.get(4));
+ assertEquals(true, row.get(5));
+ assertEquals("value", row.get(6));
+
+ var schema = parser.buildConfiguredEventSchema(columns, "timestamp");
+ var runtimeTypes = schema.getEventProperties().stream()
+ .map(property -> ((EventPropertyPrimitive) property).getRuntimeType())
+ .toList();
+
+ assertEquals(List.of(
+ XSD.LONG.toString(),
+ XSD.INTEGER.toString(),
+ XSD.LONG.toString(),
+ XSD.FLOAT.toString(),
+ XSD.DOUBLE.toString(),
+ XSD.BOOLEAN.toString(),
+ XSD.STRING.toString()
+ ), runtimeTypes);
+ }
+
+ @Test
+ void shouldRejectValuesThatDoNotMatchConfiguredTypes() {
+ var request = new CsvImportRequest();
+ request.setCsvConfig(makeConfig(",", ".", true));
+ request.setTimestampColumn("timestamp");
+ request.setColumns(List.of(
+ makeColumn("timestamp", "timestamp", "LONG"),
+ makeColumn("integerValue", "integerValue", "INTEGER"),
+ makeColumn("booleanValue", "booleanValue", "BOOLEAN")
+ ));
+ request.setRows(List.of(List.of("1710000000000", "17.5", "not-a-boolean")));
+
+ var exception = assertThrows(CsvImportValidationException.class, () -> parser.toImportRows(request));
+
+ assertTrue(exception.getValidationMessages().get(0).getMessage().contains("integerValue"));
+
+ request.setRows(List.of(List.of("1710000000000", "17.0", "not-a-boolean")));
+
+ exception = assertThrows(CsvImportValidationException.class, () -> parser.toImportRows(request));
+
+ assertTrue(exception.getValidationMessages().get(0).getMessage().contains("booleanValue"));
+ }
+
private CsvImportConfiguration makeConfig(String delimiter, String decimalSeparator, boolean hasHeader) {
var config = new CsvImportConfiguration();
config.setDelimiter(delimiter);
diff --git a/ui/cypress/support/utils/dataset/DatasetSeedUtils.ts b/ui/cypress/support/utils/dataset/DatasetSeedUtils.ts
index ca370519f9..92dfdab0c6 100644
--- a/ui/cypress/support/utils/dataset/DatasetSeedUtils.ts
+++ b/ui/cypress/support/utils/dataset/DatasetSeedUtils.ts
@@ -18,7 +18,13 @@
import * as CSV from 'csv-string';
-type CsvRuntimeType = 'STRING' | 'BOOLEAN' | 'LONG' | 'FLOAT';
+type CsvRuntimeType =
+ | 'STRING'
+ | 'BOOLEAN'
+ | 'INTEGER'
+ | 'LONG'
+ | 'FLOAT'
+ | 'DOUBLE';
type CsvImportTargetMode = 'NEW' | 'EXISTING';
type CsvImportJobState = 'RUNNING' | 'SUCCEEDED' | 'FAILED';
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 2bcc5bc441..be2bdbc948 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
@@ -20,7 +20,13 @@ import { EventSchema } from '../gen/streampipes-model';
export type CsvImportTargetMode = 'NEW' | 'EXISTING';
export type CsvImportJobState = 'RUNNING' | 'SUCCEEDED' | 'FAILED';
-export type CsvRuntimeType = 'STRING' | 'BOOLEAN' | 'LONG' | 'FLOAT';
+export type CsvRuntimeType =
+ | 'STRING'
+ | 'BOOLEAN'
+ | 'INTEGER'
+ | 'LONG'
+ | 'FLOAT'
+ | 'DOUBLE';
export interface CsvImportConfiguration {
delimiter: string;
diff --git a/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.spec.ts b/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.spec.ts
new file mode 100644
index 0000000000..b71ee3e5db
--- /dev/null
+++ b/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-dialog.component.spec.ts
@@ -0,0 +1,55 @@
+/*
+ * 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 '@angular/compiler';
+import { CsvRuntimeType, DataType } from '@streampipes/platform-services';
+import { describe, expect, it } from 'vitest';
+import { CsvImportDialogComponent } from './csv-import-dialog.component';
+
+interface CsvImportTypeMappings {
+ toSupportedColumnType(type: string): CsvRuntimeType;
+ toRuntimeType(type: string): string;
+ fromRuntimeType(type: string): CsvRuntimeType;
+}
+
+describe('CsvImportDialogComponent type mappings', () => {
+ const component = Object.create(
+ CsvImportDialogComponent.prototype,
+ ) as CsvImportTypeMappings;
+ const runtimeTypes: Array<[CsvRuntimeType, string]> = [
+ ['STRING', DataType.STRING],
+ ['BOOLEAN', DataType.BOOLEAN],
+ ['INTEGER', DataType.INTEGER],
+ ['LONG', DataType.LONG],
+ ['FLOAT', DataType.FLOAT],
+ ['DOUBLE', DataType.DOUBLE],
+ ];
+
+ it('preserves every supported CSV type for existing datasets', () => {
+ runtimeTypes.forEach(([csvType]) => {
+ expect(component.toSupportedColumnType(csvType)).toBe(csvType);
+ });
+ });
+
+ it('maps every supported CSV type in both directions', () => {
+ runtimeTypes.forEach(([csvType, xsdType]) => {
+ expect(component.toRuntimeType(csvType)).toBe(xsdType);
+ expect(component.fromRuntimeType(xsdType)).toBe(csvType);
+ });
+ });
+});
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 16593e169e..9baf515e96 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
@@ -587,9 +587,12 @@ export class CsvImportDialogComponent {
}
private toColumnModel(column: CsvImportColumn): CsvImportColumnModel {
- const initialType = this.toDefaultColumnType(
- column.runtimeType || column.inferredType || 'STRING',
- );
+ const sourceType =
+ column.runtimeType || column.inferredType || 'STRING';
+ const initialType =
+ this.targetMode() === 'EXISTING'
+ ? this.toSupportedColumnType(sourceType)
+ : this.toDefaultColumnType(sourceType);
const property = new EventPropertyPrimitive();
property['@class'] =
'org.apache.streampipes.model.schema.EventPropertyPrimitive';
@@ -637,31 +640,35 @@ export class CsvImportDialogComponent {
switch (type) {
case 'BOOLEAN':
return DataType.BOOLEAN;
+ case 'INTEGER':
+ return DataType.INTEGER;
case 'LONG':
return DataType.LONG;
case 'FLOAT':
return DataType.FLOAT;
+ case 'DOUBLE':
+ return DataType.DOUBLE;
default:
return DataType.STRING;
}
}
- private fromRuntimeType(
- type: string,
- ): 'STRING' | 'BOOLEAN' | 'LONG' | 'FLOAT' {
+ private fromRuntimeType(type: string): CsvRuntimeType {
if (type === DataType.BOOLEAN) {
return 'BOOLEAN';
- } else if (type === DataType.LONG || type === DataType.INTEGER) {
+ } else if (type === DataType.INTEGER) {
+ return 'INTEGER';
+ } else if (type === DataType.LONG) {
return 'LONG';
- } else if (type === DataType.FLOAT || type === DataType.DOUBLE) {
+ } else if (type === DataType.FLOAT) {
return 'FLOAT';
+ } else if (type === DataType.DOUBLE) {
+ return 'DOUBLE';
}
return 'STRING';
}
- private toDefaultColumnType(
- type: string,
- ): 'STRING' | 'BOOLEAN' | 'LONG' | 'FLOAT' {
+ private toDefaultColumnType(type: string): CsvRuntimeType {
if (type === 'BOOLEAN') {
return 'BOOLEAN';
} else if (type === 'STRING') {
@@ -670,6 +677,19 @@ export class CsvImportDialogComponent {
return 'FLOAT';
}
+ private toSupportedColumnType(type: string): CsvRuntimeType {
+ switch (type) {
+ case 'BOOLEAN':
+ case 'INTEGER':
+ case 'LONG':
+ case 'FLOAT':
+ case 'DOUBLE':
+ return type;
+ default:
+ return 'STRING';
+ }
+ }
+
private invalidatePreview(): void {
if (this.previewReloadTimeout) {
clearTimeout(this.previewReloadTimeout);
diff --git a/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-preview-table/csv-import-preview-table.component.html b/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-preview-table/csv-import-preview-table.component.html
index 335fd01f74..1636b30a9d 100644
--- a/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-preview-table/csv-import-preview-table.component.html
+++ b/ui/src/app/dataset/dialog/csv-import-dialog/csv-import-preview-table/csv-import-preview-table.component.html
@@ -113,12 +113,18 @@
<mat-option value="BOOLEAN">
BOOLEAN
</mat-option>
+ <mat-option value="INTEGER">
+ INTEGER
+ </mat-option>
<mat-option value="LONG">
LONG
</mat-option>
<mat-option value="FLOAT">
FLOAT
</mat-option>
+ <mat-option value="DOUBLE">
+ DOUBLE
+ </mat-option>
</mat-select>
</mat-form-field>
}