DRILL-7978: Fixed Width Format Plugin

contrib/format-fixedwidth/pom.xml

@@ -0,0 +1,85 @@
+<?xml version="1.0"?>
+<!--
+
+    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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <artifactId>drill-contrib-parent</artifactId>
+    <groupId>org.apache.drill.contrib</groupId>
+    <version>2.0.0-SNAPSHOT</version>
+  </parent>
+  <artifactId>drill-format-fixedwidth</artifactId>
+  <name>Drill : Contrib : Format : FixedWidth</name>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.drill.exec</groupId>
+        <artifactId>drill-java-exec</artifactId>
+        <version>${project.version}</version>
+    </dependency>
+<!--    <dependency>-->
+<!--      <groupId>com.epam</groupId>-->
+<!--      <artifactId>parso</artifactId>-->
+<!--      <version>2.0.14</version>-->
+<!--    </dependency>-->
+
+    <!-- Test dependencies -->
+    <dependency>
+      <groupId>org.apache.drill.exec</groupId>
+      <artifactId>drill-java-exec</artifactId>
+      <classifier>tests</classifier>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.drill</groupId>
+      <artifactId>drill-common</artifactId>
+      <classifier>tests</classifier>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+  <build>
+    <plugins>
+      <plugin>
+        <artifactId>maven-resources-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>copy-java-sources</id>
+            <phase>process-sources</phase>
+            <goals>
+              <goal>copy-resources</goal>
+            </goals>
+            <configuration>
+              <outputDirectory>${basedir}/target/classes/org/apache/drill/exec/store/fixedwidth</outputDirectory>
+              <resources>
+                <resource>
+                  <directory>src/main/java/org/apache/drill/exec/store/fixedwidth</directory>
+                  <filtering>true</filtering>
+                </resource>
+              </resources>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+
+</project>

contrib/format-fixedwidth/src/main/java/org/apache/drill/exec/store/fixedwidth/FixedWidthBatchReader.java

@@ -0,0 +1,99 @@
+package org.apache.drill.exec.store.fixedwidth;
+
+import org.apache.drill.common.AutoCloseables;
+import org.apache.drill.common.exceptions.CustomErrorContext;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.types.TypeProtos;
+import org.apache.drill.exec.physical.impl.scan.v3.ManagedReader;
+import org.apache.drill.exec.physical.impl.scan.v3.file.FileSchemaNegotiator;
+import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
+import org.apache.drill.exec.record.metadata.SchemaBuilder;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.shaded.guava.com.google.common.base.Charsets;
+import org.apache.hadoop.mapred.FileSplit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+//import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+
+public class FixedWidthBatchReader implements ManagedReader {
+
+  private final int maxRecords;  // Do we need this?
+  private final FixedWidthFormatConfig config;
+  private InputStream fsStream;
+  private ResultSetLoader loader;
+  private FileSplit split;
+  private CustomErrorContext errorContext;
+  private static final Logger logger = LoggerFactory.getLogger(FixedWidthBatchReader.class);
+  private BufferedReader reader;
+
+  public FixedWidthBatchReader(FileSchemaNegotiator negotiator, FixedWidthFormatConfig config, int maxRecords) {
+    this.loader = open(negotiator);
+    this.config = config;
+    this.maxRecords = maxRecords;
+  }
+
+  @Override
+  public boolean next() {
+    return true;
+  }
+
+  @Override
+  public void close() {
+    if (fsStream != null){
+      AutoCloseables.closeSilently(fsStream);
+      fsStream = null;
+    }
+  }
+
+  private ResultSetLoader open(FileSchemaNegotiator negotiator) {
+//    this.split = (FileSplit) negotiator.split();
+    this.errorContext = negotiator.parentErrorContext();
+//    openFile(negotiator);
+
+    try {
+      negotiator.tableSchema(buildSchema(), true);
+      this.loader = negotiator.build();
+    } catch (Exception e) {
+      throw UserException
+        .dataReadError(e)
+        .message("Failed to open input file: {}", this.split.getPath().toString())
+        .addContext(this.errorContext)
+        .addContext(e.getMessage())
+        .build(FixedWidthBatchReader.logger);
+    }
+    this.reader = new BufferedReader(new InputStreamReader(this.fsStream, Charsets.UTF_8));
+    return this.loader;
+  }
+
+//  private void openFile(FileSchemaNegotiator negotiator) {
+//    try {
+//      this.fsStream = negotiator.file().fileSystem().openPossiblyCompressedStream(this.split.getPath());
+//      sasFileReader = new SasFileReaderImpl(this.fsStream);
+//      firstRow = sasFileReader.readNext();
+//    } catch (IOException e) {
+//      throw UserException
+//        .dataReadError(e)
+//        .message("Unable to open Fixed Width File %s", this.split.getPath())
+//        .addContext(e.getMessage())
+//        .addContext(this.errorContext)
+//        .build(FixedWidthBatchReader.logger);
+//    }
+//  }
+
+  private TupleMetadata buildSchema() {
+    SchemaBuilder builder = new SchemaBuilder();
+    for (FixedWidthFieldConfig field : config.getFields()) {
+      if (field.getType() == TypeProtos.MinorType.VARDECIMAL){
+        builder.addNullable(field.getName(), TypeProtos.MinorType.VARDECIMAL,38,4);
+        //revisit this
+      } else {
+        builder.addNullable(field.getName(), field.getType());
+      }
+    }
+    return builder.buildSchema();
+  }
+}

contrib/format-fixedwidth/src/main/java/org/apache/drill/exec/store/fixedwidth/FixedWidthFieldConfig.java

@@ -0,0 +1,111 @@
+/*
+ * 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.drill.exec.store.fixedwidth;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.drill.common.PlanStringBuilder;
+import org.apache.drill.common.types.TypeProtos;
+
+import java.util.Objects;
+
+
+@JsonTypeName("fixedwidthReaderFieldDescription")
+@JsonInclude(JsonInclude.Include.NON_DEFAULT)
+public class FixedWidthFieldConfig implements Comparable<FixedWidthFieldConfig> {
+
+  private final String name;
+  private final int index;
+  private int width;
+  private TypeProtos.MinorType type;
+  private final String dateTimeFormat;
+
+  public FixedWidthFieldConfig(@JsonProperty("name") String name,
+                               @JsonProperty("index") int index,
+                               @JsonProperty("width") int width,
+                               @JsonProperty("type") TypeProtos.MinorType type) {
+    this(name, index, width, type, null);
+  }
+
+  @JsonCreator
+  public FixedWidthFieldConfig(@JsonProperty("name") String name,
+                               @JsonProperty("index") int index,
+                               @JsonProperty("width") int width,
+                               @JsonProperty("type") TypeProtos.MinorType type,
+                               @JsonProperty("dateTimeFormat") String dateTimeFormat) {
+    this.name = name;
+    this.index = index;
+    this.width = width;
+    this.type = type;
+    this.dateTimeFormat = dateTimeFormat;
+  }
+
+  public String getName() {return name;}
+
+  public int getIndex() {return index;}
+
+  public int getWidth() {return width;}
+
+  public TypeProtos.MinorType getType() {return type;}
+
+  public void setType() {
+    this.type = TypeProtos.MinorType.VARCHAR;
+  }
+
+  public String getDateTimeFormat() {return dateTimeFormat;}
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(name, index, width, type,  dateTimeFormat);
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj) {
+      return true;
+    }
+    if (obj == null || getClass() != obj.getClass()) {
+      return false;
+    }
+    FixedWidthFieldConfig other = (FixedWidthFieldConfig) obj;
+    return Objects.equals(name, other.name)
+      && Objects.equals(index, other.index)
+      && Objects.equals(width, other.width)
+      && Objects.equals(type, other.type)
+      && Objects.equals(dateTimeFormat, other.dateTimeFormat);
+  }
+
+  @Override
+  public String toString() {
+    return new PlanStringBuilder(this)
+      .field("name", name)
+      .field("index", index)
+      .field("width", width)
+      .field("type", type)
+      .field("dateTimeFormat", dateTimeFormat)
+      .toString();
+  }
+
+  @Override
+  public int compareTo(FixedWidthFieldConfig o) {
+    return Integer.compare(this.getIndex(), o.getIndex());
+  }
+}