Skip to content

Commit b69eb43

Browse files
vikasrathee-csvikasrathee-cs
authored
Merge pull request #136 from cloudsufi/add-filter-pr
Support for Filter Queries in Split class.
2 parents a11a825 + c57f884 commit b69eb43

17 files changed

+149
-107
lines changed

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

Lines changed: 37 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -122,24 +122,26 @@ public String getAccessTokenRetryableMode() throws ExecutionException, RetryExce
122122
}
123123

124124
/**
125-
* Fetch the list of records from ServiceNow table.
125+
* Fetches a list of records from a ServiceNow table.
126126
*
127-
* @param tableName The ServiceNow table name
128-
* @param valueType The value type
129-
* @param startDate The start date
130-
* @param endDate The end date
131-
* @param offset The number of records to skip
132-
* @param limit The number of records to be fetched
133-
* @return The list of Map; each Map representing a table row
127+
* @param tableName The name of the ServiceNow table.
128+
* @param valueType The type of value to retrieve (e.g., actual or display).
129+
* @param filterQuery An encoded query to filter the records. For example, to filter records created
130+
* between two dates, the query would be: {@code
131+
* sys_created_onBETWEENjavascript:gs.dateGenerate
132+
* ('2023-01-01','start')@javascript:gs.dateGenerate('2023-01-31','end')}
133+
* @param offset The number of records to skip from the beginning of the result set.
134+
* @param limit The maximum number of records to retrieve.
135+
* @return A list of maps, where each map represents a record from the table.
136+
* @throws ServiceNowAPIException If an error occurs while fetching the records.
134137
*/
135138
public List<Map<String, String>> fetchTableRecords(
136-
String tableName,
137-
SourceValueType valueType,
138-
String startDate,
139-
String endDate,
140-
int offset,
141-
int limit)
142-
throws ServiceNowAPIException {
139+
String tableName,
140+
SourceValueType valueType,
141+
String filterQuery,
142+
int offset,
143+
int limit)
144+
throws ServiceNowAPIException {
143145
ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
144146
this.conf.getRestApiEndpoint(), tableName, false, schemaType)
145147
.setExcludeReferenceLink(true)
@@ -150,39 +152,16 @@ public List<Map<String, String>> fetchTableRecords(
150152
requestBuilder.setOffset(offset);
151153
}
152154

153-
applyDateRangeToRequest(requestBuilder, startDate, endDate);
155+
if (!Util.isNullOrEmpty(filterQuery)) {
156+
requestBuilder.setQuery(filterQuery);
157+
}
154158

155159
String accessToken = getAccessToken();
156160
requestBuilder.setAuthHeader(accessToken);
157161
RestAPIResponse apiResponse = executeGetWithRetries(requestBuilder.build());
158162
return parseResponseToResultListOfMap(apiResponse.getResponseBody());
159163
}
160164

161-
private void applyDateRangeToRequest(ServiceNowTableAPIRequestBuilder requestBuilder, String startDate,
162-
String endDate) {
163-
String dateRange = generateDateRangeQuery(startDate, endDate);
164-
if (!Strings.isNullOrEmpty(dateRange)) {
165-
requestBuilder.setQuery(dateRange);
166-
}
167-
}
168-
169-
private String generateDateRangeQuery(String startDate, String endDate) {
170-
if (Util.isNullOrEmpty(startDate) || Util.isNullOrEmpty(endDate)) {
171-
return "";
172-
}
173-
174-
String dateRange = "";
175-
try {
176-
String createdOnDateRange = String.format(DATE_RANGE_TEMPLATE, FIELD_CREATED_ON, startDate, endDate);
177-
String updatedOnDateRange = String.format(DATE_RANGE_TEMPLATE, FIELD_UPDATED_ON, startDate, endDate);
178-
dateRange = String.format("%s^OR%s", createdOnDateRange, updatedOnDateRange);
179-
} catch (Exception e) {
180-
LOG.error("Error in generateDateRangeQuery, hence ignoring the date range", e);
181-
}
182-
183-
return dateRange;
184-
}
185-
186165
private int getRecordCountFromHeader(RestAPIResponse apiResponse) {
187166
String headerValue = apiResponse.getHeaders().get(ServiceNowConstants.HEADER_NAME_TOTAL_COUNT);
188167
return Strings.isNullOrEmpty(headerValue) ? 0 : Integer.parseInt(headerValue);
@@ -226,18 +205,16 @@ private String getErrorMessage(String responseBody) {
226205
*
227206
* @param tableName The ServiceNow table name
228207
* @param valueType The value type
229-
* @param startDate The start date
230-
* @param endDate The end date
231208
* @param offset The number of records to skip
232209
* @param limit The number of records to be fetched
233210
* @return The list of Map; each Map representing a table row
234211
*/
235212
public List<Map<String, String>> fetchTableRecordsRetryableMode(String tableName, SourceValueType valueType,
236-
String startDate, String endDate, int offset,
213+
String filterQuery, int offset,
237214
int limit) throws ServiceNowAPIException {
238215
final List<Map<String, String>> results = new ArrayList<>();
239216
Callable<Boolean> fetchRecords = () -> {
240-
results.addAll(fetchTableRecords(tableName, valueType, startDate, endDate, offset, limit));
217+
results.addAll(fetchTableRecords(tableName, valueType, filterQuery, offset, limit));
241218
return true;
242219
};
243220

@@ -415,15 +392,18 @@ private Schema prepareSchemaWithMetadataAPI(RestAPIResponse restAPIResponse, Lis
415392
}
416393

417394
/**
418-
* Get the total number of records in the table
395+
* Gets the total number of records in the table based on the provided date range.
419396
*
420-
* @param tableName ServiceNow table name for which record count is fetched.
421-
* @return the table record count
422-
* @throws ServiceNowAPIException
397+
* @param tableName ServiceNow table name for which record count is fetched
398+
* @param startDate Start date to use for filtering records
399+
* @param endDate End date to use for filtering records
400+
* @return The total number of records
401+
* @throws ServiceNowAPIException If an error occurs while fetching the record count
423402
*/
424-
public int getTableRecordCount(String tableName)
403+
public int getTableRecordCount(String tableName, @Nullable String startDate, @Nullable String endDate)
425404
throws ServiceNowAPIException {
426-
return getTableRecordCount(tableName, getAccessToken());
405+
String filterQuery = Util.generateDateRangeQuery(startDate, endDate);
406+
return getTableRecordCountUsingFilterQuery(tableName, getAccessToken(), filterQuery);
427407
}
428408

429409
/**
@@ -434,13 +414,18 @@ public int getTableRecordCount(String tableName)
434414
* @return the table record count
435415
* @throws ServiceNowAPIException
436416
*/
437-
public int getTableRecordCount(String tableName, String accessToken) throws ServiceNowAPIException {
417+
public int getTableRecordCountUsingFilterQuery(String tableName, String accessToken,
418+
@Nullable String filterQuery)
419+
throws ServiceNowAPIException {
438420
ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
439421
this.conf.getRestApiEndpoint(), tableName, false, schemaType)
440422
.setExcludeReferenceLink(true)
441423
.setDisplayValue(SourceValueType.SHOW_DISPLAY_VALUE)
442424
.setLimit(1);
443425
RestAPIResponse apiResponse = null;
426+
if (!Util.isNullOrEmpty(filterQuery)) {
427+
requestBuilder.setQuery(filterQuery);
428+
}
444429
requestBuilder.setResponseHeaders(ServiceNowConstants.HEADER_NAME_TOTAL_COUNT);
445430
requestBuilder.setAuthHeader(accessToken);
446431
apiResponse = executeGetWithRetries(requestBuilder.build());

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -214,7 +214,9 @@ private Schema getSchema(String tableName) {
214214
tableName,
215215
null,
216216
SourceValueType.SHOW_DISPLAY_VALUE,
217-
true);
217+
true,
218+
null,
219+
null);
218220
Schema schema = tableInfo.stream().findFirst().isPresent() ? tableInfo.stream().findFirst().get().getSchema() :
219221
null;
220222
return schema;

‎src/main/java/io/cdap/plugin/servicenow/source/ServiceNowInputFormat.java‎

Lines changed: 16 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import io.cdap.plugin.servicenow.util.SourceApplication;
2626
import io.cdap.plugin.servicenow.util.SourceQueryMode;
2727
import io.cdap.plugin.servicenow.util.SourceValueType;
28+
import io.cdap.plugin.servicenow.util.Util;
2829
import org.apache.hadoop.conf.Configuration;
2930
import org.apache.hadoop.io.NullWritable;
3031
import org.apache.hadoop.mapreduce.InputFormat;
@@ -64,7 +65,8 @@ public static List<ServiceNowTableInfo> setInput(Configuration jobConfig, Source
6465
// return the schema object for each table as ServiceNowTableInfo
6566
List<ServiceNowTableInfo> tableInfos = fetchTableInfo(mode, conf.getConnection(), conf.getTableName(),
6667
conf.getApplicationName(), conf.getValueType(),
67-
conf.getUseConnection());
68+
conf.getUseConnection(), conf.getStartDate(),
69+
conf.getEndDate());
6870
jobConf.setTableInfos(tableInfos);
6971

7072
return tableInfos;
@@ -74,10 +76,11 @@ public static List<ServiceNowTableInfo> fetchTableInfo(SourceQueryMode mode, Ser
7476
@Nullable String tableName,
7577
@Nullable SourceApplication application,
7678
@Nullable SourceValueType valueType,
77-
@Nullable Boolean useConnection) {
79+
@Nullable Boolean useConnection,
80+
@Nullable String startdate, @Nullable String enddate) {
7881
// When mode = Table, fetch details from the table name provided in plugin config
7982
if (mode == SourceQueryMode.TABLE) {
80-
ServiceNowTableInfo tableInfo = getTableMetaData(tableName, conf, valueType, useConnection);
83+
ServiceNowTableInfo tableInfo = getTableMetaData(tableName, conf, valueType, useConnection, startdate, enddate);
8184
return (tableInfo == null) ? Collections.emptyList() : Collections.singletonList(tableInfo);
8285
}
8386

@@ -87,7 +90,7 @@ public static List<ServiceNowTableInfo> fetchTableInfo(SourceQueryMode mode, Ser
8790

8891
List<String> tableNames = application.getTableNames();
8992
for (String table : tableNames) {
90-
ServiceNowTableInfo tableInfo = getTableMetaData(table, conf, valueType, useConnection);
93+
ServiceNowTableInfo tableInfo = getTableMetaData(table, conf, valueType, useConnection, startdate, enddate);
9194
if (tableInfo == null) {
9295
continue;
9396
}
@@ -100,7 +103,8 @@ public static List<ServiceNowTableInfo> fetchTableInfo(SourceQueryMode mode, Ser
100103
private static ServiceNowTableInfo getTableMetaData(String tableName,
101104
ServiceNowConnectorConfig conf,
102105
SourceValueType valueType,
103-
@Nullable Boolean useConnection) {
106+
@Nullable Boolean useConnection, @Nullable String startdate,
107+
@Nullable String enddate) {
104108
// Call API to fetch first record from the table
105109
ServiceNowTableAPIClientImpl restApi = new ServiceNowTableAPIClientImpl(conf, useConnection);
106110

@@ -111,7 +115,7 @@ private static ServiceNowTableInfo getTableMetaData(String tableName,
111115
if (schema == null) {
112116
return null;
113117
}
114-
recordCount = restApi.getTableRecordCount(tableName);
118+
recordCount = restApi.getTableRecordCount(tableName, startdate, enddate);
115119
} catch (ServiceNowAPIException e) {
116120
throw new RuntimeException(String.format("Error in fetching table metadata due to reason: %s", e.getMessage()),
117121
e);
@@ -132,6 +136,10 @@ public List<InputSplit> getSplits(JobContext jobContext) {
132136
*/
133137
public List<InputSplit> getSplits(Configuration configuration) {
134138
ServiceNowJobConfiguration jobConfig = new ServiceNowJobConfiguration(configuration);
139+
String startdate = jobConfig.getPluginConf().getStartDate();
140+
String enddate = jobConfig.getPluginConf().getEndDate();
141+
String filterQuery = Util.generateDateRangeQuery(startdate, enddate);
142+
135143
int pageSize = jobConfig.getPluginConf().getPageSize().intValue();
136144
List<ServiceNowTableInfo> tableInfos = jobConfig.getTableInfos();
137145
List<InputSplit> resultSplits = new ArrayList<>();
@@ -141,7 +149,7 @@ public List<InputSplit> getSplits(Configuration configuration) {
141149
int totalRecords = tableInfo.getRecordCount();
142150
if (totalRecords <= pageSize) {
143151
// add single split for table and continue
144-
resultSplits.add(new ServiceNowInputSplit(tableName, 0));
152+
resultSplits.add(new ServiceNowInputSplit(tableName, 0, filterQuery));
145153
continue;
146154
}
147155

@@ -152,7 +160,7 @@ public List<InputSplit> getSplits(Configuration configuration) {
152160
int offset = 0;
153161

154162
for (int page = 1; page <= pages; page++) {
155-
resultSplits.add(new ServiceNowInputSplit(tableName, offset));
163+
resultSplits.add(new ServiceNowInputSplit(tableName, offset, filterQuery));
156164
offset += pageSize;
157165
}
158166
}

‎src/main/java/io/cdap/plugin/servicenow/source/ServiceNowInputSplit.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,14 +29,16 @@
2929
public class ServiceNowInputSplit extends InputSplit implements Writable {
3030
private String tableName;
3131
private int offset;
32+
private String filterQuery;
3233

3334
// used by mapreduce
3435
public ServiceNowInputSplit() {
3536
}
3637

37-
public ServiceNowInputSplit(String tableName, int offset) {
38+
public ServiceNowInputSplit(String tableName, int offset, String filterQuery) {
3839
this.tableName = tableName;
3940
this.offset = offset;
41+
this.filterQuery = filterQuery;
4042
}
4143

4244
public String getTableName() {
@@ -47,16 +49,22 @@ public int getOffset() {
4749
return offset;
4850
}
4951

52+
public String getFilterQuery() {
53+
return filterQuery;
54+
}
55+
5056
@Override
5157
public void write(DataOutput dataOutput) throws IOException {
5258
dataOutput.writeUTF(this.tableName);
5359
dataOutput.writeInt(this.offset);
60+
dataOutput.writeUTF(this.filterQuery);
5461
}
5562

5663
@Override
5764
public void readFields(DataInput dataInput) throws IOException {
5865
this.tableName = dataInput.readUTF();
5966
this.offset = dataInput.readInt();
67+
this.filterQuery = dataInput.readUTF();
6068
}
6169

6270
@Override

‎src/main/java/io/cdap/plugin/servicenow/source/ServiceNowMultiInputFormat.java‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import io.cdap.plugin.servicenow.util.ServiceNowConstants;
2626
import io.cdap.plugin.servicenow.util.ServiceNowTableInfo;
2727
import io.cdap.plugin.servicenow.util.SourceValueType;
28+
import io.cdap.plugin.servicenow.util.Util;
2829
import org.apache.hadoop.conf.Configuration;
2930
import org.apache.hadoop.io.NullWritable;
3031
import org.apache.hadoop.mapreduce.InputFormat;
@@ -45,6 +46,7 @@
4546
import java.util.Set;
4647
import java.util.stream.Collectors;
4748
import java.util.stream.Stream;
49+
import javax.annotation.Nullable;
4850

4951
/**
5052
* ServiceNow input format.
@@ -68,21 +70,23 @@ public static Set<ServiceNowTableInfo> setInput(Configuration jobConfig,
6870
// Depending on conf value fetch the list of fields for each table and create schema object
6971
// return the schema object for each table as ServiceNowTableInfo
7072
Set<ServiceNowTableInfo> tableInfos = fetchTablesInfo(conf.getConnection(), conf.getTableNames(),
71-
conf.getUseConnection());
73+
conf.getUseConnection(), conf.getStartDate(),
74+
conf.getEndDate());
7275

7376
jobConf.setTableInfos(tableInfos.stream().collect(Collectors.toList()));
7477

7578
return tableInfos;
7679
}
7780

7881
static Set<ServiceNowTableInfo> fetchTablesInfo(ServiceNowConnectorConfig conf, String tableNames,
79-
Boolean useConnection) {
82+
Boolean useConnection, @Nullable String startdate,
83+
@Nullable String enddate) {
8084

8185
Set<ServiceNowTableInfo> tablesInfos = new LinkedHashSet<>();
8286

8387
Set<String> tableNameSet = getList(tableNames);
8488
for (String table : tableNameSet) {
85-
ServiceNowTableInfo tableInfo = getTableMetaData(table, conf, useConnection);
89+
ServiceNowTableInfo tableInfo = getTableMetaData(table, conf, useConnection, startdate, enddate);
8690
if (tableInfo == null) {
8791
continue;
8892
}
@@ -93,7 +97,8 @@ static Set<ServiceNowTableInfo> fetchTablesInfo(ServiceNowConnectorConfig conf,
9397
}
9498

9599
private static ServiceNowTableInfo getTableMetaData(String tableName, ServiceNowConnectorConfig conf,
96-
Boolean useConnection) {
100+
Boolean useConnection, @Nullable String startdate,
101+
@Nullable String enddate) {
97102
// Call API to fetch first record from the table
98103
ServiceNowTableAPIClientImpl restApi = new ServiceNowTableAPIClientImpl(conf, useConnection);
99104

@@ -107,7 +112,7 @@ private static ServiceNowTableInfo getTableMetaData(String tableName, ServiceNow
107112
if (schema == null) {
108113
return null;
109114
}
110-
recordCount = restApi.getTableRecordCount(tableName);
115+
recordCount = restApi.getTableRecordCount(tableName, startdate, enddate);
111116
} catch (ServiceNowAPIException e) {
112117
throw new RuntimeException(e);
113118
}
@@ -127,6 +132,11 @@ public static Set<String> getList(String value) {
127132
@Override
128133
public List<InputSplit> getSplits(JobContext jobContext) throws IOException, InterruptedException {
129134
ServiceNowJobConfiguration jobConfig = new ServiceNowJobConfiguration(jobContext.getConfiguration());
135+
136+
String startdate = jobConfig.getMultiSourcePluginConf().getStartDate();
137+
String enddate = jobConfig.getMultiSourcePluginConf().getEndDate();
138+
String filterQuery = Util.generateDateRangeQuery(startdate, enddate);
139+
130140
int pageSize = jobConfig.getPluginConf().getPageSize().intValue();
131141
List<ServiceNowTableInfo> tableInfos = jobConfig.getTableInfos();
132142
List<InputSplit> resultSplits = new ArrayList<>();
@@ -142,7 +152,7 @@ public List<InputSplit> getSplits(JobContext jobContext) throws IOException, Int
142152
int offset = 0;
143153

144154
for (int page = 1; page <= pages; page++) {
145-
resultSplits.add(new ServiceNowInputSplit(tableName, offset));
155+
resultSplits.add(new ServiceNowInputSplit(tableName, offset, filterQuery));
146156
offset += pageSize;
147157
}
148158
}

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

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -95,9 +95,8 @@ public StructuredRecord getCurrentValue() throws IOException {
9595
void fetchData() throws ServiceNowAPIException {
9696
// Get the table data
9797
results = restApi.fetchTableRecordsRetryableMode(tableName, multiSourcePluginConf.getValueType(),
98-
multiSourcePluginConf.getStartDate(),
99-
multiSourcePluginConf.getEndDate(), split.getOffset(),
100-
multiSourcePluginConf.getPageSize());
98+
split.getFilterQuery(), split.getOffset(),
99+
multiSourcePluginConf.getPageSize());
101100

102101
iterator = results.iterator();
103102
}

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -107,8 +107,8 @@ public StructuredRecord getCurrentValue() throws IOException {
107107

108108
private void fetchData() throws ServiceNowAPIException {
109109
// Get the table data
110-
results = restApi.fetchTableRecordsRetryableMode(tableName, pluginConf.getValueType(), pluginConf.getStartDate(),
111-
pluginConf.getEndDate(), split.getOffset(),
110+
results = restApi.fetchTableRecordsRetryableMode(tableName, pluginConf.getValueType(), split.getFilterQuery(),
111+
split.getOffset(),
112112
pluginConf.getPageSize());
113113
LOG.debug("Results size={}", results.size());
114114

0 commit comments

Comments
 (0)