Merge pull request #132 from cloudsufi/feature/PLUGIN-1940

PLUGIN-1940 : Support Array Type

Author: sgarg-CS

src/e2e-test/java/io/cdap/plugin/servicenowsink/actions/ServiceNowSinkPropertiesPageActions.java

@@ -60,7 +60,7 @@ public static void getRecordFromServiceNowTable(String query, String tableName)
         System.getenv("SERVICE_NOW_REST_API_ENDPOINT"),
         System.getenv("SERVICE_NOW_USERNAME"),
         System.getenv("SERVICE_NOW_PASSWORD"),
-        "", "", "", null);
+        "", "", "", null, false);
 
     ServiceNowTableAPIClientImpl tableAPIClient = new ServiceNowTableAPIClientImpl(config.getConnection(), true);
     responseFromServiceNowTable = tableAPIClient.getRecordFromServiceNowTable(tableName, query);

src/e2e-test/java/io/cdap/plugin/tests/hooks/TestSetupHooks.java

@@ -59,7 +59,7 @@ public static void initializeServiceNowSourceConfig() {
         System.getenv("SERVICE_NOW_REST_API_ENDPOINT"),
         System.getenv("SERVICE_NOW_USERNAME"),
         System.getenv("SERVICE_NOW_PASSWORD"),
-        "", "", "", null);
+        "", "", "", null, false);
   }
 
   @Before(order = 2, value = "@SN_PRODUCT_CATALOG_ITEM")

src/main/java/io/cdap/plugin/servicenow/apiclient/ServiceNowTableAPIClientImpl.java

@@ -27,6 +27,7 @@
 import com.google.gson.JsonArray;
 import com.google.gson.JsonObject;
 import com.google.gson.reflect.TypeToken;
+import com.sun.org.apache.xpath.internal.operations.Bool;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.plugin.servicenow.connector.ServiceNowConnectorConfig;
@@ -291,7 +292,7 @@ public MetadataAPISchemaResponse parseSchemaResponse(String responseBody) {
    */
   public Schema fetchTableSchema(String tableName, SourceValueType valueType)
       throws ServiceNowAPIException {
-    return fetchTableSchema(tableName, getAccessToken(), valueType, schemaType);
+    return fetchTableSchema(tableName, getAccessToken(), valueType, schemaType, false);
   }
 
   private SchemaType getSchemaTypeBasedOnUseConnection(Boolean useConnection) {
@@ -310,10 +311,11 @@ private SchemaType getSchemaTypeBasedOnUseConnection(Boolean useConnection) {
    * @param accessToken Access Token to use
    * @param valueType Type of value (Actual/Display)
    * @param schemaType Enum to determine which approach to take to fetch schema.
+   * @param enableNewDataTypes Flag to enable new data types
    * @return schema for given ServiceNow table
    */
   public Schema fetchTableSchema(String tableName, String accessToken, SourceValueType valueType,
-                                 SchemaType schemaType)
+                                 SchemaType schemaType, Boolean enableNewDataTypes)
       throws ServiceNowAPIException {
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
       this.conf.getRestApiEndpoint(), tableName, true, schemaType)
@@ -325,7 +327,7 @@ public Schema fetchTableSchema(String tableName, String accessToken, SourceValue
     List<ServiceNowColumn> columns = new ArrayList<>();
 
     if (schemaType == SchemaType.METADATA_API_BASED) {
-      return prepareSchemaWithMetadataAPI(restAPIResponse, columns, tableName, valueType);
+      return prepareSchemaWithMetadataAPI(restAPIResponse, columns, tableName, valueType, enableNewDataTypes);
     } else if (schemaType == SchemaType.SCHEMA_API_BASED) {
       return prepareSchemaWithSchemaAPI(restAPIResponse, columns, tableName);
     } else {
@@ -362,7 +364,7 @@ private Schema prepareSchemaWithSchemaAPI(RestAPIResponse restAPIResponse, List<
     for (SchemaAPISchemaField field : schemaAPISchemaResponse.getResult()) {
       columns.add(new ServiceNowColumn(field.getName(), field.getInternalType()));
     }
-    return SchemaBuilder.constructSchema(tableName, columns);
+    return SchemaBuilder.constructSchema(tableName, columns, false);
   }
 
   /**
@@ -384,8 +386,7 @@ private Schema prepareSchemaWithSchemaAPI(RestAPIResponse restAPIResponse, List<
    * @throws RuntimeException if the response does not contain valid column information.
    */
   private Schema prepareSchemaWithMetadataAPI(RestAPIResponse restAPIResponse, List<ServiceNowColumn> columns,
-                                              String tableName, SourceValueType valueType) throws
-    ServiceNowAPIException {
+    String tableName, SourceValueType valueType, Boolean enableNewDataTypes) throws ServiceNowAPIException {
     MetadataAPISchemaResponse metadataAPISchemaResponse = parseSchemaResponse(restAPIResponse.getResponseBody());
 
     if (metadataAPISchemaResponse.getResult() == null || metadataAPISchemaResponse.getResult().getColumns() == null ||
@@ -410,7 +411,7 @@ private Schema prepareSchemaWithMetadataAPI(RestAPIResponse restAPIResponse, Lis
         columns.add(new ServiceNowColumn(field.getName(), field.getInternalType()));
       }
     }
-    return SchemaBuilder.constructSchema(tableName, columns);
+    return SchemaBuilder.constructSchema(tableName, columns, enableNewDataTypes);
   }
 
   /**
@@ -552,7 +553,7 @@ private Schema prepareStringBasedSchema(RestAPIResponse restAPIResponse, List<Se
       for (String key : firstRecord.keySet()) {
         columns.add(new ServiceNowColumn(key, "string"));
       }
-      return SchemaBuilder.constructSchema(tableName, columns);
+      return SchemaBuilder.constructSchema(tableName, columns, false);
     }
     return null;
   }

src/main/java/io/cdap/plugin/servicenow/connector/ServiceNowConnector.java

@@ -193,7 +193,7 @@ private List<StructuredRecord> getTableData(String tableName, int limit)
         StructuredRecord.Builder recordBuilder = StructuredRecord.builder(schema);
         for (Schema.Field field : tableFields) {
           String fieldName = field.getName();
-          ServiceNowRecordConverter.convertToValue(fieldName, field.getSchema(), result.get(i), recordBuilder);
+          ServiceNowRecordConverter.convertToValue(fieldName, field.getSchema(), result.get(i), recordBuilder, false);
         }
         StructuredRecord structuredRecord = recordBuilder.build();
         recordList.add(structuredRecord);

src/main/java/io/cdap/plugin/servicenow/connector/ServiceNowRecordConverter.java

@@ -32,6 +32,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * Utility class for converting the record from ServiceNow data type to CDAP schema data types
@@ -83,7 +84,7 @@ public class ServiceNowRecordConverter {
     ));
 
   public static void convertToValue(String fieldName, Schema fieldSchema, Map<String, String> record,
-                                    StructuredRecord.Builder recordBuilder) {
+    StructuredRecord.Builder recordBuilder, Boolean enableNewDataTypes) {
     String fieldValue = record.get(fieldName);
     if (fieldValue == null || fieldValue.isEmpty()) {
       // Set 'null' value as it is
@@ -128,13 +129,26 @@ public static void convertToValue(String fieldName, Schema fieldSchema, Map<Stri
       case BOOLEAN:
         recordBuilder.set(fieldName, convertToBooleanValue(fieldValue));
         return;
+      case ARRAY:
+        recordBuilder.set(fieldName, enableNewDataTypes.equals(Boolean.TRUE) ? convertToList(fieldValue) : fieldValue);
+        return;
       default:
         throw new IllegalStateException(
           String.format("Record type '%s' is not supported for field '%s'", fieldType.name(), fieldName));
     }
 
   }
 
+  public static List<String> convertToList(String fieldValue) {
+    if (fieldValue.trim().isEmpty()) {
+      return Collections.emptyList();
+    }
+    return Arrays.stream(fieldValue.split(","))
+      .map(String::trim)
+      .filter(s -> !s.isEmpty())
+      .collect(Collectors.toList());
+  }
+
   @VisibleForTesting
   public static Double convertToDoubleValue(String fieldValue) {
     try {

src/main/java/io/cdap/plugin/servicenow/source/ServiceNowBaseSourceConfig.java

@@ -74,6 +74,11 @@ public class ServiceNowBaseSourceConfig extends ServiceNowBaseConfig {
   @Nullable
   @Description("The number of records to fetch from ServiceNow. Default is 5000.")
   private Integer pageSize;
+  
+  @Name(ServiceNowConstants.PROPERTY_ENABLE_NEW_DATA_TYPES)
+  @Nullable
+  @Description("Enable support for new data types such as array (glide_list) ")
+  private Boolean enableNewDataTypes;
 
   /**
    * Constructor for ServiceNowSourceConfig object.
@@ -92,14 +97,16 @@ public class ServiceNowBaseSourceConfig extends ServiceNowBaseConfig {
    */
   public ServiceNowBaseSourceConfig(String referenceName, String clientId, String clientSecret, String restApiEndpoint,
                                     String user, String password, String tableNameField, String valueType,
-                                    @Nullable String startDate, @Nullable String endDate, Integer pageSize) {
+                                    @Nullable String startDate, @Nullable String endDate, Integer pageSize,
+                                    Boolean enableNewDataTypes) {
     super(clientId, clientSecret, restApiEndpoint, user, password);
     this.referenceName = referenceName;
     this.tableNameField = tableNameField;
     this.valueType = valueType;
     this.startDate = startDate;
     this.endDate = endDate;
     this.pageSize = pageSize;
+    this.enableNewDataTypes = enableNewDataTypes;
   }
 
   public String getReferenceName() {
@@ -124,6 +131,10 @@ public Integer getPageSize() {
     return pageSize == null ? ServiceNowConstants.PAGE_SIZE : pageSize;
   }
 
+  public Boolean getEnableNewDataTypes() {
+    return enableNewDataTypes == null ? Boolean.FALSE : enableNewDataTypes;
+  }
+
   /**
    * Validates {@link ServiceNowBaseSourceConfig} instance.
    */

src/main/java/io/cdap/plugin/servicenow/source/ServiceNowMultiRecordReader.java

@@ -36,6 +36,7 @@ public class ServiceNowMultiRecordReader extends ServiceNowBaseRecordReader {
 
   private final ServiceNowMultiSourceConfig multiSourcePluginConf;
   private ServiceNowTableAPIClientImpl restApi;
+  private Boolean enableNewDataTypes;
 
   ServiceNowMultiRecordReader(ServiceNowMultiSourceConfig multiSourcePluginConf) {
     super();
@@ -81,8 +82,8 @@ public StructuredRecord getCurrentValue() throws IOException {
     try {
       for (Schema.Field field : tableFields) {
         String fieldName = field.getName();
-        ServiceNowRecordConverter.convertToValue(fieldName, field.getSchema(), row,
-                                                 recordBuilder);
+        ServiceNowRecordConverter.convertToValue(fieldName, field.getSchema(), row, recordBuilder,
+          multiSourcePluginConf.getEnableNewDataTypes());
       }
     } catch (Exception e) {
       throw new IOException("Error decoding row from table " + tableName, e);

src/main/java/io/cdap/plugin/servicenow/source/ServiceNowMultiSourceConfig.java

@@ -55,9 +55,9 @@ public class ServiceNowMultiSourceConfig extends ServiceNowBaseSourceConfig {
   public ServiceNowMultiSourceConfig(String referenceName, String clientId, String clientSecret, String restApiEndpoint,
                                      String user, String password, String tableNameField, String valueType,
                                      @Nullable String startDate, @Nullable String endDate, Integer pageSize,
-                                     String tableNames) {
+                                     String tableNames, Boolean enableNewDataTypes) {
     super(referenceName, clientId, clientSecret, restApiEndpoint, user, password, tableNameField, valueType, startDate,
-          endDate, pageSize);
+          endDate, pageSize, enableNewDataTypes);
     this.tableNames = tableNames;
   }
 

src/main/java/io/cdap/plugin/servicenow/source/ServiceNowRecordReader.java

@@ -38,6 +38,7 @@ public class ServiceNowRecordReader extends ServiceNowBaseRecordReader {
   private static final Logger LOG = LoggerFactory.getLogger(ServiceNowRecordReader.class);
   private final ServiceNowSourceConfig pluginConf;
   private ServiceNowTableAPIClientImpl restApi;
+  private Boolean enableNewDataTypes;
 
   public ServiceNowRecordReader(ServiceNowSourceConfig pluginConf) {
     super();
@@ -94,7 +95,8 @@ public StructuredRecord getCurrentValue() throws IOException {
     try {
       for (Schema.Field field : tableFields) {
         String fieldName = field.getName();
-        ServiceNowRecordConverter.convertToValue(fieldName, field.getSchema(), row, recordBuilder);
+        ServiceNowRecordConverter.convertToValue(fieldName, field.getSchema(), row, recordBuilder,
+          pluginConf.getEnableNewDataTypes());
       }
     } catch (Exception e) {
       LOG.error("Error decoding row from table " + tableName, e);

src/main/java/io/cdap/plugin/servicenow/source/ServiceNowSourceConfig.java

@@ -79,9 +79,9 @@ public ServiceNowSourceConfig(String referenceName, String queryMode, @Nullable
                                 @Nullable String tableNameField, @Nullable String tableName, String clientId,
                                 String clientSecret, String restApiEndpoint, String user, String password,
                                 String valueType, @Nullable String startDate, @Nullable String endDate,
-                                Integer pageSize) {
+                                Integer pageSize, Boolean enableNewDataTypes) {
     super(referenceName, clientId, clientSecret, restApiEndpoint, user, password, tableNameField, valueType, startDate,
-          endDate, pageSize);
+          endDate, pageSize, enableNewDataTypes);
     this.referenceName = referenceName;
     this.queryMode = queryMode;
     this.applicationName = applicationName;