diff --git a/wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/calcite/converter/WayangTableScanVisitor.java b/wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/calcite/converter/WayangTableScanVisitor.java index 569384697..e6ec4085c 100755 --- a/wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/calcite/converter/WayangTableScanVisitor.java +++ b/wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/calcite/converter/WayangTableScanVisitor.java @@ -18,12 +18,12 @@ package org.apache.wayang.api.sql.calcite.converter; -import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rel.type.RelDataTypeField; import org.apache.wayang.api.sql.calcite.rel.WayangTableScan; import org.apache.wayang.api.sql.calcite.utils.ModelParser; -import org.apache.wayang.api.sql.sources.fs.JavaCSVTableSource; +import org.apache.wayang.java.operators.JavaCSVFileSource; +import org.apache.wayang.java.operators.CsvType; import org.apache.wayang.core.plan.wayangplan.Operator; import org.apache.wayang.core.types.DataSetType; import org.apache.wayang.jdbc.operators.JdbcTableSource; @@ -64,15 +64,31 @@ Operator visit(final WayangTableScan wayangRelNode) { "Could not initialize calcite model parser from current Wayang configuration"); } - final List fieldTypes = wayangRelNode.getRowType().getFieldList().stream() + final List fieldTypes = wayangRelNode.getRowType().getFieldList().stream() .map(RelDataTypeField::getType) + .map(type -> { + switch (type.getSqlTypeName()) { + case BOOLEAN: return CsvType.BOOLEAN; + case TINYINT: return CsvType.TINYINT; + case SMALLINT: return CsvType.SMALLINT; + case INTEGER: return CsvType.INTEGER; + case BIGINT: return CsvType.BIGINT; + case FLOAT: return CsvType.FLOAT; + case DOUBLE: return CsvType.DOUBLE; + case DECIMAL: return CsvType.decimal(type.getPrecision(), type.getScale()); + case DATE: return CsvType.DATE; + case TIME: return CsvType.TIME; + case TIMESTAMP: return CsvType.TIMESTAMP; + default: return CsvType.STRING; + } + }) .collect(Collectors.toList()); final String url = String.format("file:/%s/%s.csv", modelParser.getFsPath(), wayangRelNode.getTableName()); final char separator = modelParser.getSchemaDelimiter(tableSource); - return new JavaCSVTableSource<>(url, DataSetType.createDefault(Record.class), fieldTypes, separator); + return new JavaCSVFileSource<>(url, DataSetType.createDefault(Record.class), fieldTypes, separator); } else if (wayangRelNode.getTable().getQualifiedName().size() == 1) { // we assume that it is coming from a test environement or in memory db. diff --git a/wayang-platforms/wayang-java/pom.xml b/wayang-platforms/wayang-java/pom.xml index d07ef27d5..3700ee815 100644 --- a/wayang-platforms/wayang-java/pom.xml +++ b/wayang-platforms/wayang-java/pom.xml @@ -38,6 +38,11 @@ + + net.sf.opencsv + opencsv + 2.3 + org.apache.wayang wayang-core diff --git a/wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/sources/fs/CsvRowConverter.java b/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/CsvRowConverter.java old mode 100755 new mode 100644 similarity index 92% rename from wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/sources/fs/CsvRowConverter.java rename to wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/CsvRowConverter.java index af490acb1..d46f0c76a --- a/wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/sources/fs/CsvRowConverter.java +++ b/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/CsvRowConverter.java @@ -15,11 +15,9 @@ * limitations under the License. */ -package org.apache.wayang.api.sql.sources.fs; +package org.apache.wayang.java.operators; import au.com.bytecode.opencsv.CSVParser; -import org.apache.calcite.avatica.util.DateTimeUtils; -import org.apache.calcite.rel.type.RelDataType; import org.apache.commons.lang3.time.FastDateFormat; import java.io.IOException; @@ -57,11 +55,11 @@ public class CsvRowConverter { - public static Object convert(RelDataType fieldType, String string) { + public static Object convert(CsvType fieldType, String string) { if (fieldType == null || string == null) { return string; } - switch (fieldType.getSqlTypeName()) { + switch (fieldType.getKind()) { case BOOLEAN: if (string.length() == 0) { return null; @@ -101,14 +99,15 @@ public static Object convert(RelDataType fieldType, String string) { if (string.length() == 0) { return null; } - return parseDecimal(fieldType.getPrecision(), fieldType.getScale(), string); + return fieldType.getPrecision() < 0 ? new BigDecimal(string) + : parseDecimal(fieldType.getPrecision(), fieldType.getScale(), string); case DATE: if (string.length() == 0) { return null; } try { Date date = TIME_FORMAT_DATE.parse(string); - return (int) (date.getTime() / DateTimeUtils.MILLIS_PER_DAY); + return (int) (date.getTime() / 86_400_000L); } catch (ParseException e) { return null; } @@ -132,8 +131,7 @@ public static Object convert(RelDataType fieldType, String string) { } catch (ParseException e) { return null; } - case GEOMETRY: - case VARCHAR: + case STRING: default: return string; } diff --git a/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/CsvType.java b/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/CsvType.java new file mode 100644 index 000000000..04c445786 --- /dev/null +++ b/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/CsvType.java @@ -0,0 +1,70 @@ +/* + * 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. + */ + +package org.apache.wayang.java.operators; + +import java.io.Serializable; + +/** CSV field type, including optional decimal precision and scale, without SQL API dependencies. */ +public final class CsvType implements Serializable { + + private static final long serialVersionUID = 1L; + + public enum Kind { + BOOLEAN, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DECIMAL, DATE, TIME, TIMESTAMP, STRING + } + + public static final CsvType BOOLEAN = new CsvType(Kind.BOOLEAN, -1, -1); + public static final CsvType TINYINT = new CsvType(Kind.TINYINT, -1, -1); + public static final CsvType SMALLINT = new CsvType(Kind.SMALLINT, -1, -1); + public static final CsvType INTEGER = new CsvType(Kind.INTEGER, -1, -1); + public static final CsvType BIGINT = new CsvType(Kind.BIGINT, -1, -1); + public static final CsvType FLOAT = new CsvType(Kind.FLOAT, -1, -1); + public static final CsvType DOUBLE = new CsvType(Kind.DOUBLE, -1, -1); + public static final CsvType DECIMAL = new CsvType(Kind.DECIMAL, -1, -1); + public static final CsvType DATE = new CsvType(Kind.DATE, -1, -1); + public static final CsvType TIME = new CsvType(Kind.TIME, -1, -1); + public static final CsvType TIMESTAMP = new CsvType(Kind.TIMESTAMP, -1, -1); + public static final CsvType STRING = new CsvType(Kind.STRING, -1, -1); + + private final Kind kind; + private final int precision; + private final int scale; + + private CsvType(Kind kind, int precision, int scale) { + this.kind = kind; + this.precision = precision; + this.scale = scale; + } + + /** Creates a decimal type with the precision and scale declared by the source schema. */ + public static CsvType decimal(int precision, int scale) { + return new CsvType(Kind.DECIMAL, precision, scale); + } + + public Kind getKind() { + return this.kind; + } + + public int getPrecision() { + return this.precision; + } + + public int getScale() { + return this.scale; + } +} diff --git a/wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/sources/fs/JavaCSVTableSource.java b/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/JavaCSVFileSource.java old mode 100755 new mode 100644 similarity index 93% rename from wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/sources/fs/JavaCSVTableSource.java rename to wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/JavaCSVFileSource.java index e37b236e8..f046b86dc --- a/wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/sources/fs/JavaCSVTableSource.java +++ b/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/JavaCSVFileSource.java @@ -15,9 +15,7 @@ * limitations under the License. */ -package org.apache.wayang.api.sql.sources.fs; - -import org.apache.calcite.rel.type.RelDataType; +package org.apache.wayang.java.operators; import org.apache.commons.io.IOUtils; @@ -37,7 +35,6 @@ import org.apache.wayang.core.util.fs.FileUtils; import org.apache.wayang.java.channels.StreamChannel; import org.apache.wayang.java.execution.JavaExecutor; -import org.apache.wayang.java.operators.JavaExecutionOperator; import java.io.BufferedReader; import java.io.IOException; @@ -51,11 +48,11 @@ import java.util.stream.Stream; import java.util.stream.StreamSupport; -public class JavaCSVTableSource extends UnarySource implements JavaExecutionOperator { +public class JavaCSVFileSource extends UnarySource implements JavaExecutionOperator { private final String sourcePath; - private final List fieldTypes; + private final List fieldTypes; private final char separator; // Default separator // TODO: incorporate fields later for projectable table scans @@ -63,12 +60,12 @@ public class JavaCSVTableSource extends UnarySource implements JavaExecuti /** * Table source with default seperator ';'

- * See {@link #JavaCSVTableSource(String, DataSetType, List, char)} for custom seperator + * See {@link #JavaCSVFileSource(String, DataSetType, List, char)} for custom seperator * @param sourcePath * @param type * @param fieldTypes */ - public JavaCSVTableSource(final String sourcePath, final DataSetType type, final List fieldTypes) { + public JavaCSVFileSource(final String sourcePath, final DataSetType type, final List fieldTypes) { super(type); this.sourcePath = sourcePath; this.fieldTypes = fieldTypes; @@ -83,7 +80,7 @@ public JavaCSVTableSource(final String sourcePath, final DataSetType type, fi * @param fieldTypes * @param separator */ - public JavaCSVTableSource(final String sourcePath, final DataSetType type, final List fieldTypes, + public JavaCSVFileSource(final String sourcePath, final DataSetType type, final List fieldTypes, final char separator) { super(type); this.sourcePath = sourcePath; diff --git a/wayang-platforms/wayang-java/src/test/java/org/apache/wayang/java/operators/JavaCSVFileSourceTest.java b/wayang-platforms/wayang-java/src/test/java/org/apache/wayang/java/operators/JavaCSVFileSourceTest.java new file mode 100644 index 000000000..9a0f59df7 --- /dev/null +++ b/wayang-platforms/wayang-java/src/test/java/org/apache/wayang/java/operators/JavaCSVFileSourceTest.java @@ -0,0 +1,83 @@ +/* + * 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. + */ + +package org.apache.wayang.java.operators; + +import org.apache.wayang.basic.data.Record; +import org.apache.wayang.core.types.DataSetType; +import org.apache.wayang.java.channels.JavaChannelInstance; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.math.BigDecimal; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +class JavaCSVFileSourceTest extends JavaExecutionOperatorTestBase { + + @TempDir + Path directory; + + private List read(String csv, List types, char separator) throws Exception { + Path file = directory.resolve("input.csv"); + Files.writeString(file, csv); + JavaCSVFileSource source = new JavaCSVFileSource<>( + file.toUri().toString(), DataSetType.createDefault(Record.class), types, separator); + JavaChannelInstance[] outputs = {createStreamChannelInstance()}; + evaluate(source, new JavaChannelInstance[0], outputs); + try (Stream records = outputs[0].provideStream()) { + return records.collect(Collectors.toList()); + } + } + + @Test + void readsTypedRecordsWithCustomSeparatorAndQuotedFields() throws Exception { + List records = read("id:int,name:string,price:decimal\n1;\"a;b\";12.345\n", + List.of(CsvType.INTEGER, CsvType.STRING, CsvType.decimal(5, 2)), ';'); + assertEquals(List.of(new Record(1, "a;b", new BigDecimal("12.35"))), records); + } + + @Test + void rejectsDecimalOverflowAfterRounding() { + assertThrows(IllegalArgumentException.class, () -> read("price:decimal\n99.995\n", + List.of(CsvType.decimal(4, 2)), ';')); + } + + @Test + void rejectsRowsWithMissingFields() { + assertThrows(IllegalStateException.class, () -> read("id:int,name:string\n1\n", + List.of(CsvType.INTEGER, CsvType.STRING), ';')); + } + + @Test + void retainsTypedHeaderValidation() { + assertThrows(IllegalStateException.class, () -> read("id,name\n1;Alice\n", + List.of(CsvType.INTEGER, CsvType.STRING), ';')); + } + + @Test + void rejectsEmptyFiles() { + assertThrows(IllegalStateException.class, () -> read("", List.of(CsvType.INTEGER), ';')); + } +}