Fixed issue with output schema in PLUGIN-409

Author: flakrimjusufi

src/main/java/io/cdap/plugin/gcp/bigquery/sink/AbstractBigQuerySink.java

@@ -333,14 +333,14 @@ private Schema getTableSchema(
     return BigQueryUtil.getTableSchema(bqSchema, collector);
   }
 
-  protected void validateInsertSchema(Table table, Schema tableSchema, FailureCollector collector) {
+  protected void validateInsertSchema(Table table, @Nullable Schema tableSchema, FailureCollector collector) {
     com.google.cloud.bigquery.Schema bqSchema = table.getDefinition().getSchema();
     if (bqSchema == null || bqSchema.getFields().isEmpty()) {
       // Table is created without schema, so no further validation is required.
       return;
     }
 
-    if (getConfig().isTruncateTableSet()) {
+    if (getConfig().isTruncateTableSet() || tableSchema == null) {
       //no validation required for schema if truncate table is set.
       // BQ will overwrite the schema for normal tables when write disposition is WRITE_TRUNCATE
       //note - If write to single partition is supported in future, schema validation will be necessary
@@ -389,10 +389,10 @@ protected void validateInsertSchema(Table table, Schema tableSchema, FailureColl
   protected void validateSchema(
     String tableName,
     com.google.cloud.bigquery.Schema bqSchema,
-    Schema tableSchema,
+    @Nullable Schema tableSchema,
     boolean allowSchemaRelaxation,
     FailureCollector collector) {
-    if (bqSchema == null || bqSchema.getFields().isEmpty()) {
+    if (bqSchema == null || bqSchema.getFields().isEmpty() || tableSchema == null) {
       // Table is created without schema, so no further validation is required.
       return;
     }

src/test/java/io/cdap/plugin/gcp/bigquery/sink/BigQuerySinkTest.java

@@ -230,6 +230,24 @@ public void testSchemaValidationAllowSchemaRelaxationNoTruncateException() throw
     Assert.assertEquals(0, collector.getValidationFailures().size());
   }
 
+  @Test
+  public void testSchemaValidationNullInputSchema() throws NoSuchFieldException {
+    BigQuerySink sink = getValidationTestSink(false);
+    MockFailureCollector collector = new MockFailureCollector("bqsink");
+    Table table = getTestSchema();
+    sink.validateSchema(
+      table.getTableId().getTable(),
+      table.getDefinition().getSchema(),
+      null,
+      true,
+      collector);
+    sink.validateInsertSchema(
+      table,
+      null,
+      collector);
+    Assert.assertEquals(0, collector.getValidationFailures().size());
+  }
+
   private Table getTestSchema() {
     Table table = mock(Table.class);
     TableId tableId = TableId.of("test", "testds", "testtab");