Merge pull request #75 from cloudsufi/feat/AbstractFileBatchSource-ui

[PLUGIN-1906] Add AbstractFileBatchSource GDrive [UI + Docs]

Author: psainics

docs/GoogleDrive-batchsource.md

@@ -33,6 +33,57 @@ Then the File Identifier would be `17W3vOhBwe0i24OdVNsbz8rAMClzUitKeAbumTqWFrkow
 They will be a part of output structured record. Descriptions for properties can be view at 
 [Drive API file reference](https://developers.google.com/drive/api/v3/reference/files).
 
+**Structured Schema Required :** Set this property to `true` if the output should be structured records.
+Using structured records allows you to use read files with formats like avro, csv, json, text, xls, etc.
+
+**Format:** Format of the data to read.
+The format must be one of 'avro', 'blob', 'csv', 'delimited', 'json', 'parquet', 'text', 'tsv', 'xls', or the
+name of any format plugin that you have deployed to your environment.
+If the format is a macro, only the pre-packaged formats can be used.
+If the format is 'blob', every input file will be read into a separate record.
+The 'blob' format also requires a schema that contains a field named 'body' of type 'bytes'.
+If the format is 'text', the schema must contain a field named 'body' of type 'string'.
+
+**Get Schema:** Auto-detects schema from file. Supported formats are: avro, parquet, csv, delimited, tsv, blob, xls
+and text.
+
+Blob - is set by default as field named 'body' of type bytes.
+
+Text - is set by default as two fields: 'body' of type bytes and 'offset' of type 'long'.
+
+JSON - is not supported, user has to manually provide the output schema.
+
+Parquet - If the path is a directory, the plugin will look for files ending in '.parquet' to read the schema from.
+If no such file can be found, an error will be returned.
+
+Avro - If the path is a directory, the plugin will look for files ending in '.avro' to read the schema from.
+If no such file can be found, an error will be returned.
+
+**Sample Size:** The maximum number of rows that will get investigated for automatic data type detection.
+The default value is 1000. This is used when the format is `xls`, `csv`, `tsv`, `delimited`.
+
+**Override:** A list of columns with the corresponding data types for whom the automatic data type detection gets
+skipped. This is used when the format is `xls`, `csv`, `tsv`, `delimited`.
+
+**Delimiter:** Delimiter to use when the format is 'delimited'. This will be ignored for other formats.
+
+**Enable Quoted Values** Whether to treat content between quotes as a value. This value will only be used if the format
+is 'csv', 'tsv' or 'delimited'. For example, if this is set to true, a line that looks like `1, "a, b, c"` will output two fields.
+The first field will have `1` as its value and the second will have `a, b, c` as its value. The quote characters will be trimmed.
+The newline delimiter cannot be within quotes.
+
+It also assumes the quotes are well enclosed. The left quote will match the first following quote right before the delimiter. If there is an
+unenclosed quote, an error will occur.
+
+**Use First Row as Header:** Whether to use the first line of each file as the column headers. Supported formats are 'text', 'csv', 'tsv', 'xls', 'delimited'.
+
+**Terminate Reading After Empty Row:** Specify whether to stop reading after encountering the first empty row. Defaults to false. When false the reader will read all rows in the sheet. This is only used when the format is 'xls'.
+
+**Select Sheet Using:** Select the sheet by name or number. Default is 'Sheet Number'. This is only used when the format is 'xls'.
+
+**Sheet Value:** The name/number of the sheet to read from. If not specified, the first sheet will be read.
+Sheet Numbers are 0 based, ie first sheet is 0. This is only used when the format is 'xls'.
+
 ### Filtering
 
 **Filter:** Filter that can be applied to the files in the selected directory. 
@@ -107,6 +158,10 @@ Default 0 value means unlimited. Is not applicable for files in Google formats.
 
 **Body Output Format** Output format for body of file. "Bytes" and "String" values are available.
 
+**File System Properties:** Additional properties to use with the InputFormat when reading the data.
+
+**File Encoding:** The character encoding for the file(s) to be read. The default encoding is UTF-8.
+
 ### Exporting
 
 **Google Documents Export Format:** MIME type which is used for Google Documents when converted to structured records.

pom.xml

@@ -63,6 +63,11 @@
       <artifactId>hydrator-common</artifactId>
       <version>${cdap.plugin.version}</version>
     </dependency>
+    <dependency>
+      <groupId>io.cdap.plugin</groupId>
+      <artifactId>format-common</artifactId>
+      <version>${cdap.plugin.version}</version>
+    </dependency>
     <dependency>
       <groupId>io.cdap.cdap</groupId>
       <artifactId>cdap-data-pipeline2_2.11</artifactId>

src/main/java/io/cdap/plugin/google/drive/source/GoogleDriveSourceConfig.java

@@ -23,7 +23,11 @@
 import io.cdap.cdap.api.annotation.Name;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.cdap.etl.api.FailureCollector;
+import io.cdap.cdap.etl.api.StageContext;
+import io.cdap.plugin.format.FileFormat;
+import io.cdap.plugin.format.plugin.FileSourceProperties;
 import io.cdap.plugin.google.common.GoogleFilteringSourceConfig;
+import io.cdap.plugin.google.common.IdentifierType;
 import io.cdap.plugin.google.common.ValidationResult;
 import io.cdap.plugin.google.common.exceptions.InvalidPropertyTypeException;
 import io.cdap.plugin.google.common.utils.ExportedType;
@@ -33,13 +37,14 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 
 /**
  * Configurations for Google Drive Batch Source plugin.
  */
-public class GoogleDriveSourceConfig extends GoogleFilteringSourceConfig {
+public class GoogleDriveSourceConfig extends GoogleFilteringSourceConfig implements FileSourceProperties {
   public static final String FILE_METADATA_PROPERTIES = "fileMetadataProperties";
   public static final String CONFIGURATION_PARSE_PROPERTY_NAME = "properties";
   public static final String FILE_TYPES_TO_PULL = "fileTypesToPull";
@@ -49,6 +54,9 @@ public class GoogleDriveSourceConfig extends GoogleFilteringSourceConfig {
   public static final String SHEETS_EXPORTING_FORMAT = "sheetsExportingFormat";
   public static final String DRAWINGS_EXPORTING_FORMAT = "drawingsExportingFormat";
   public static final String PRESENTATIONS_EXPORTING_FORMAT = "presentationsExportingFormat";
+  public static final String IS_STRUCTURED_SCHEMA_REQUIRED = "structuredSchemaRequired";
+  public static final String NAME_SCHEMA = "schema";
+  public static final String NAME_FORMAT = "format";
 
   public static final String DEFAULT_BODY_FORMAT = "bytes";
   public static final long DEFAULT_MAX_PARTITION_SIZE = 0;
@@ -61,6 +69,12 @@ public class GoogleDriveSourceConfig extends GoogleFilteringSourceConfig {
   public static final String FILE_TYPES_TO_PULL_LABEL = "File types to pull";
   public static final String BODY_FORMAT_LABEL = "Body output format";
 
+  public static final String GOOGLE_DRIVE_SCHEMA = "drive";
+  public static final String GOOGLE_DRIVE_AUTHORITY = "drive.google.com";
+  public static final String GOOGLE_DRIVE_FILE_PATH_PREFIX = "/drive/file/d";
+  public static final String GOOGLE_DRIVE_FOLDER_PATH_PREFIX = "/drive/folders";
+  public static final String GOOGLE_DRIVE_DEFAULT_FILENAME = "default.txt";
+
   @Nullable
   @Name(FILE_METADATA_PROPERTIES)
   @Description("Properties that represent metadata of files. \n" +
@@ -113,7 +127,91 @@ public class GoogleDriveSourceConfig extends GoogleFilteringSourceConfig {
   @Nullable
   @Macro
   protected String presentationsExportingFormat;
-  private transient Schema schema = null;
+
+  @Macro
+  @Nullable
+  @Description("Output schema for the source. Formats like 'avro' and 'parquet' require a schema in order to "
+    + "read the data.")
+  private String schema;
+
+  @Name(IS_STRUCTURED_SCHEMA_REQUIRED)
+  @Description("Wheather to fetch schema or not")
+  @Nullable
+  protected Boolean isStructuredSchemaRequired;
+
+  @Name(NAME_FORMAT)
+  @Macro
+  @Description("Format of the data to read. Supported formats are 'csv'....")
+  @Nullable
+  private String format;
+
+  @Macro
+  @Nullable
+  @Description("Whether to recursively read directories within the input directory. The default is false.")
+  private Boolean recursive;
+
+  @Macro
+  @Nullable
+  @Description("Whether to allow an input that does not exist. When false, the source will fail the run if the input "
+      + "does not exist. When true, the run will not fail and the source will not generate any output. "
+      + "The default value is false.")
+  private Boolean ignoreNonExistingFolders;
+
+  @Macro
+  @Nullable
+  @Description("The maximum number of rows that will get investigated for automatic data type detection.")
+  private Long sampleSize;
+
+  @Macro
+  @Nullable
+  @Description("A list of columns with the corresponding data types for whom the automatic data type detection gets " +
+      "skipped.")
+  private String override;
+
+  @Macro
+  @Nullable
+  @Description("The delimiter to use if the format is 'delimited'. The delimiter will be ignored if the format "
+      + "is anything other than 'delimited'.")
+  private String delimiter;
+
+  @Macro
+  @Nullable
+  @Description("Whether to use first row as header. Supported formats are 'text', 'csv', 'tsv', " +
+      "'delimited'. Default value is false.")
+  private Boolean skipHeader;
+
+  @Macro
+  @Nullable
+  @Description("Whether to treat content between quotes as a value. This value will only be used if the format " +
+      "is 'csv', 'tsv' or 'delimited'. The default value is false.")
+  protected Boolean enableQuotedValues;
+
+  @Macro
+  @Nullable
+  @Description("Any additional properties to use when reading from the filesystem. "
+      + "This is an advanced feature that requires knowledge of the properties supported by the underlying filesystem.")
+  private String fileSystemProperties;
+
+  @Macro
+  @Nullable
+  @Description("File encoding for the source files. The default encoding is 'UTF-8'")
+  private String fileEncoding;
+
+  @Macro
+  @Nullable
+  @Description("Select the sheet by name or number. Default is 'Sheet Number'.")
+  private String sheet;
+
+  @Macro
+  @Nullable
+  @Description("The name/number of the sheet to read from. If not specified, the first sheet will be read." +
+      "Sheet Numbers are 0 based, ie first sheet is 0.")
+  private String sheetValue;
+
+  @Macro
+  @Nullable
+  @Description("Specify whether to stop reading after encountering the first empty row. Defaults to false.")
+  private String terminateIfEmptyRow;
 
   public GoogleDriveSourceConfig(String referenceName, @Nullable String fileMetadataProperties, String fileTypesToPull,
                                  String maxPartitionSize, String bodyFormat, String sheetsExportingFormat,
@@ -134,15 +232,103 @@ public GoogleDriveSourceConfig(String referenceName, @Nullable String fileMetada
     this.endDate = endDate;
   }
 
+  @Override
+  public void validate(FailureCollector collector) {
+    getSchema();
+    // Extra validation when structure schema is required
+  }
+
+  @Override
+  public String getPath() {
+    IdentifierType idType = getIdentifierType();
+    if (idType == IdentifierType.FILE_IDENTIFIER) {
+      return String.format("%s://%s%s/%s/%s", GOOGLE_DRIVE_SCHEMA, GOOGLE_DRIVE_AUTHORITY,
+        GOOGLE_DRIVE_FILE_PATH_PREFIX, getFileIdentifier(), GOOGLE_DRIVE_DEFAULT_FILENAME);
+    } else if (idType == IdentifierType.DIRECTORY_IDENTIFIER) {
+      return String.format("%s://%s%s/%s/", GOOGLE_DRIVE_SCHEMA, GOOGLE_DRIVE_AUTHORITY,
+        GOOGLE_DRIVE_FOLDER_PATH_PREFIX, getDirectoryIdentifier());
+    }
+    throw new IllegalArgumentException(String.format("Invalid identifier type '%s'. Expected one of: %s or %s.", idType,
+        IdentifierType.FILE_IDENTIFIER, IdentifierType.DIRECTORY_IDENTIFIER));
+  }
+
+  @Override
+  public String getPath(StageContext context) {
+    return getPath();
+  }
+
+  @Override
+  public String getFormatName() {
+    // need to do this for backwards compatibility, where the pre-packaged format names were case insensitive.
+    try {
+      FileFormat fileFormat = FileFormat.from(format, x -> true);
+      return fileFormat.name().toLowerCase();
+    } catch (IllegalArgumentException e) {
+      // ignore
+    }
+    return format;
+  }
+
+  @Nullable
+  @Override
+  public FileFormat getFormat() {
+    throw new UnsupportedOperationException("GDrive does not support: FileFormat getFormat() method");
+  }
+
+  @Nullable
+  @Override
+  public Pattern getFilePattern() {
+    return null;
+  }
+
+  @Override
+  public long getMaxSplitSize() {
+    return Long.MAX_VALUE;
+  }
+
+  @Override
+  public boolean shouldAllowEmptyInput() {
+    return ignoreNonExistingFolders != null && ignoreNonExistingFolders;
+  }
+
+  @Override
+  public boolean shouldReadRecursively() {
+    return recursive != null && recursive;
+  }
+
+  @Nullable
+  @Override
+  public String getPathField() {
+    return null;
+  }
+
+  @Override
+  public boolean useFilenameAsPath() {
+    throw new UnsupportedOperationException("GDrive does not support: boolean useFilenameAsPath() method");
+  }
+
+  @Override
+  public boolean skipHeader() {
+    throw new UnsupportedOperationException("GDrive does not support: boolean skipHeader() method");
+  }
+
   /**
+   * throw new UnsupportedOperationException("GDrive  does not support: /**() method;
    * Returns the instance of Schema.
    * @return The instance of Schema
    */
   public Schema getSchema() {
-    if (schema == null) {
-      schema = SchemaBuilder.buildSchema(getFileMetadataProperties(), getBodyFormat());
+    if (!isStructuredSchemaRequired() && Strings.isNullOrEmpty(schema)) {
+      schema = SchemaBuilder.buildSchema(getFileMetadataProperties(), getBodyFormat()).toString();
+    }
+    if (Strings.isNullOrEmpty(schema)) {
+      return null;
+    }
+    try {
+      return Schema.parseJson(schema);
+    } catch (Exception e) {
+      throw new IllegalArgumentException("Invalid schema: " + e.getMessage(), e);
     }
-    return schema;
   }
 
   /**
@@ -236,8 +422,15 @@ public String getDrawingsExportingFormat() {
   }
 
   public String getPresentationsExportingFormat() {
-    return Strings.isNullOrEmpty(presentationsExportingFormat) ? DEFAULT_PRESENTATIONS_EXPORTING_FORMAT
-        : presentationsExportingFormat;
+    return Strings.isNullOrEmpty(presentationsExportingFormat) ?
+      DEFAULT_PRESENTATIONS_EXPORTING_FORMAT : presentationsExportingFormat;
+  }
+
+  public boolean isStructuredSchemaRequired() {
+    if (isStructuredSchemaRequired == null) {
+      return false; // for backward compatibility, default to false
+    }
+    return isStructuredSchemaRequired;
   }
 
   public GoogleDriveSourceConfig(String referenceName) {
@@ -284,8 +477,8 @@ public void setFilter(String filter) {
     this.filter = filter;
   }
 
-  public void setSchema(String schema) throws IOException {
-    this.schema = Schema.parseJson(schema);
+  public void setSchema(String schema) {
+    this.schema = schema;
   }
 
   public void setModificationDateRange(String modificationDateRange) {