Move JavaCSVTableSource to Java platform and rename to JavaCSVFileSource

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<RelDataType> fieldTypes = wayangRelNode.getRowType().getFieldList().stream()
+            final List<CsvType> 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.
 

wayang-platforms/wayang-java/pom.xml

@@ -38,6 +38,11 @@
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>net.sf.opencsv</groupId>
+            <artifactId>opencsv</artifactId>
+            <version>2.3</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.wayang</groupId>
             <artifactId>wayang-core</artifactId>

wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/sources/fs/CsvRowConverter.java → 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;
         }

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;
+    }
+}

wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/sources/fs/JavaCSVTableSource.java → 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,24 +48,24 @@
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
-public class JavaCSVTableSource<T> extends UnarySource<T> implements JavaExecutionOperator {
+public class JavaCSVFileSource<T> extends UnarySource<T> implements JavaExecutionOperator {
 
     private final String sourcePath;
 
-    private final List<RelDataType> fieldTypes;
+    private final List<CsvType> fieldTypes;
     private final char separator; // Default separator
 
     // TODO: incorporate fields later for projectable table scans
     // private final ImmutableIntList fields;
 
     /**
      * Table source with default seperator ';' <p>
-     * 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<T> type, final List<RelDataType> fieldTypes) {
+    public JavaCSVFileSource(final String sourcePath, final DataSetType<T> type, final List<CsvType> fieldTypes) {
         super(type);
         this.sourcePath = sourcePath;
         this.fieldTypes = fieldTypes;
@@ -83,7 +80,7 @@ public JavaCSVTableSource(final String sourcePath, final DataSetType<T> type, fi
      * @param fieldTypes
      * @param separator
      */
-    public JavaCSVTableSource(final String sourcePath, final DataSetType<T> type, final List<RelDataType> fieldTypes,
+    public JavaCSVFileSource(final String sourcePath, final DataSetType<T> type, final List<CsvType> fieldTypes,
             final char separator) {
         super(type);
         this.sourcePath = sourcePath;

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<Record> read(String csv, List<CsvType> types, char separator) throws Exception {
+        Path file = directory.resolve("input.csv");
+        Files.writeString(file, csv);
+        JavaCSVFileSource<Record> source = new JavaCSVFileSource<>(
+                file.toUri().toString(), DataSetType.createDefault(Record.class), types, separator);
+        JavaChannelInstance[] outputs = {createStreamChannelInstance()};
+        evaluate(source, new JavaChannelInstance[0], outputs);
+        try (Stream<Record> records = outputs[0].provideStream()) {
+            return records.collect(Collectors.toList());
+        }
+    }
+
+    @Test
+    void readsTypedRecordsWithCustomSeparatorAndQuotedFields() throws Exception {
+        List<Record> 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), ';'));
+    }
+}