PLUGIN-1883: Fix failing test case and set schema to null if both importQuery & tableName is null

Author: sgarg-CS

src/main/java/io/cdap/plugin/snowflake/common/util/SchemaHelper.java

@@ -115,7 +115,7 @@ private static Schema getParsedSchema(String schema) {
     }
     try {
       return Schema.parseJson(schema);
-    } catch (IOException e) {
+    } catch (IOException | IllegalStateException e) {
       throw new SchemaParseException(e);
     }
   }
@@ -129,7 +129,7 @@ private static Schema getSchema(SnowflakeAccessor snowflakeAccessor, @Nullable S
       } else if (!Strings.isNullOrEmpty(importQuery)) {
         result = snowflakeAccessor.describeQuery(importQuery);
       } else {
-        throw new SchemaParseException("Either 'Import Query' or 'Table Name' must be provided");
+        return null;
       }
       List<Schema.Field> fields = result.stream()
         .map(fieldDescriptor -> Schema.Field.of(fieldDescriptor.getName(), getSchema(fieldDescriptor)))

src/test/java/io/cdap/plugin/snowflake/common/util/SchemaHelperTest.java

@@ -65,7 +65,7 @@ public void testGetSchemaInvalidJson() {
   }
 
   @Test
-  public void testGetSchemaFromSnowflakeUnknownType() throws IOException {
+  public void testGetSchemaFromSnowflakeUnknownType() throws IOException, SQLException {
     String importQuery = "SELECT * FROM someTable";
     String tableName = "USER";
     MockFailureCollector collector = new MockFailureCollector(MOCK_STAGE);
@@ -74,7 +74,7 @@ public void testGetSchemaFromSnowflakeUnknownType() throws IOException {
     List<SnowflakeFieldDescriptor> sample = new ArrayList<>();
     sample.add(new SnowflakeFieldDescriptor("field1", -1000, false));
 
-    Mockito.when(snowflakeAccessor.describeQuery(importQuery)).thenReturn(sample);
+    Mockito.when(snowflakeAccessor.describeTable(null, tableName)).thenReturn(sample);
 
     SchemaHelper.getSchema(snowflakeAccessor, null, collector, tableName, importQuery);