Merge pull request #41 from data-integrations/feature/CDAP-14473-macro-enable-schema

CDAP-14473 macro enable schemas

Author: albertshau

src/main/java/co/cask/gcp/bigquery/BigQuerySinkConfig.java

@@ -30,29 +30,26 @@
  * configuring the <code>BigQuerySink</code> plugin.
  */
 public final class BigQuerySinkConfig extends GCPReferenceSinkConfig {
-  @Name("dataset")
+  @Macro
   @Description("The dataset to write to. A dataset is contained within a specific project. "
     + "Datasets are top-level containers that are used to organize and control access to tables and views.")
-  @Macro
   public String dataset;
 
-  @Name("table")
+  @Macro
   @Description("The table to write to. A table contains individual records organized in rows. "
     + "Each record is composed of columns (also called fields). "
     + "Every table is defined by a schema that describes the column names, data types, and other information.")
-  @Macro
   public String table;
 
+  @Macro
   @Nullable
-  @Name("bucket")
   @Description("The Google Cloud Storage bucket to store temporary data in. "
     + "It will be automatically created if it does not exist, but will not be automatically deleted. "
     + "Cloud Storage data will be deleted after it is loaded into BigQuery. " +
     "If it is not provided, a unique bucket will be created and then deleted after the run finishes.")
-  @Macro
   public String bucket;
 
-  @Name("schema")
+  @Macro
   @Description("The schema of the data to write. Must be compatible with the table schema.")
   public String schema;
 

src/main/java/co/cask/gcp/bigquery/BigQuerySourceConfig.java

@@ -29,31 +29,27 @@
  * Holds configuration required for configuring {@link BigQuerySource}.
  */
 public final class BigQuerySourceConfig extends GCPReferenceSourceConfig {
-  @Name("dataset")
+  @Macro
   @Description("The dataset the table belongs to. A dataset is contained within a specific project. "
     + "Datasets are top-level containers that are used to organize and control access to tables and views.")
-  @Macro
   public String dataset;
 
-  @Name("table")
+  @Macro
   @Description("The table to read from. A table contains individual records organized in rows. "
     + "Each record is composed of columns (also called fields). "
     + "Every table is defined by a schema that describes the column names, data types, and other information.")
-  @Macro
   public String table;
 
+  @Macro
   @Nullable
-  @Name("bucket")
   @Description("The Google Cloud Storage bucket to store temporary data in. "
     + "It will be automatically created if it does not exist, but will not be automatically deleted. "
     + "Temporary data will be deleted after it has been read. " +
     "If it is not provided, a unique bucket will be created and then deleted after the run finishes.")
-  @Macro
   public String bucket;
 
-  @Name("schema")
-  @Description("The schema of the table to read.")
   @Macro
+  @Description("The schema of the table to read.")
   public String schema;
 
   /**

src/main/java/co/cask/gcp/gcs/source/GCSSource.java

@@ -122,6 +122,7 @@ public static class GCSSourceConfig extends GCPReferenceSourceConfig implements
       + "'parquet', 'text', and 'tsv'.")
     private String format;
 
+    @Macro
     @Nullable
     @Description("Output schema. If a Path Field is set, it must be present in the schema as a string.")
     private String schema;

src/main/java/co/cask/gcp/spanner/sink/SpannerSinkConfig.java

@@ -60,7 +60,9 @@ public class SpannerSinkConfig extends GCPReferenceSinkConfig {
 
   public void validate() {
     super.validate();
-    SpannerUtil.validateSchema(getSchema());
+    if (!containsMacro("schema")) {
+      SpannerUtil.validateSchema(getSchema());
+    }
     if (!containsMacro("batchSize") && batchSize != null && batchSize < 1) {
       throw new IllegalArgumentException("Spanner batch size for writes should be positive");
     }

src/main/java/co/cask/gcp/spanner/source/SpannerSourceConfig.java

@@ -59,7 +59,9 @@ public class SpannerSourceConfig extends GCPReferenceSourceConfig {
 
   public void validate() {
     super.validate();
-    SpannerUtil.validateSchema(getSchema());
+    if (!containsMacro("schema")) {
+      SpannerUtil.validateSchema(getSchema());
+    }
     if (!containsMacro("maxPartitions") && maxPartitions != null && maxPartitions < 1) {
       throw new IllegalArgumentException("Max partitions should be positive");
     }