Skip to content

Commit 84dd41f

Browse files
author
hknokh2
committed
fix: normalize Bulk Query null values by field metadata
1 parent 6e44c3d commit 84dd41f

4 files changed

Lines changed: 193 additions & 40 deletions

File tree

‎src/modules/common/Common.ts‎

Lines changed: 123 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -804,6 +804,38 @@ export class Common {
804804
.replace(new RegExp(`[${CONSTANTS.COMPLEX_FIELDS_QUERY_SEPARATOR}]`, 'g'), CONSTANTS.COMPLEX_FIELDS_SEPARATOR);
805805
}
806806

807+
/**
808+
* Normalizes CSV-like record values using the same null-token and type casting rules as CSV input.
809+
*
810+
* @param records - Records to normalize.
811+
* @param columnToColumnDataTypeMap - Optional column-to-type mapping.
812+
* @param preserveRawNullTokens - Keeps null tokens as raw string values when true.
813+
* @returns Normalized record copies.
814+
*/
815+
public static normalizeCsvRecordValues(
816+
records: Array<Record<string, unknown>>,
817+
columnToColumnDataTypeMap?: Map<string, string>,
818+
preserveRawNullTokens = false
819+
): Array<Record<string, unknown>> {
820+
if (!records.length) {
821+
return records;
822+
}
823+
const normalizedColumnDataTypeMap = Common._normalizeCsvColumnDataTypeMap(columnToColumnDataTypeMap);
824+
return records.map((record) => {
825+
const normalized: Record<string, unknown> = {};
826+
Object.keys(record).forEach((columnName) => {
827+
normalized[columnName] = Common._normalizeCsvCellValue(
828+
record[columnName],
829+
columnName,
830+
columnToColumnDataTypeMap,
831+
normalizedColumnDataTypeMap,
832+
preserveRawNullTokens
833+
);
834+
});
835+
return normalized;
836+
});
837+
}
838+
807839
/**
808840
* Reads CSV file from the disk.
809841
*
@@ -824,57 +856,27 @@ export class Common {
824856
const readDelimiter = useInternalCsvFormat ? Common.INTERNAL_CSV_FILE_DELIMITER : Common.csvReadFileDelimiter;
825857
const readEncoding = useInternalCsvFormat ? Common.INTERNAL_CSV_FILE_ENCODING : Common.csvFileEncoding;
826858
const shouldStripUtf8Bom = useInternalCsvFormat || (Common.csvUseUtf8Bom && Common._isUtf8Encoding(readEncoding));
827-
const normalizedColumnDataTypeMap = new Map<string, string>();
828-
columnToColumnDataTypeMap?.forEach((fieldType, columnName) => {
829-
const normalizedName = columnName.trim().toLowerCase();
830-
if (!normalizedName) {
831-
return;
832-
}
833-
if (!normalizedColumnDataTypeMap.has(normalizedName)) {
834-
normalizedColumnDataTypeMap.set(normalizedName, fieldType);
835-
}
836-
});
859+
const normalizedColumnDataTypeMap = Common._normalizeCsvColumnDataTypeMap(columnToColumnDataTypeMap);
837860
function csvCast(value: string, context: CastingContext): unknown {
838861
if (context.header || typeof context.column === 'undefined') {
839862
return value;
840863
}
841864

842-
const rawValue = typeof value === 'string' ? value : String(value ?? '');
843-
const normalizedRawValue = rawValue.replace(/\r+$/g, '');
844-
const trimmedValue = normalizedRawValue.trim();
845-
if (!trimmedValue.length) {
846-
if (preserveRawNullTokens) {
847-
return normalizedRawValue;
848-
}
849-
return Common.csvInsertNulls ? null : '';
850-
}
851-
852-
if (!preserveRawNullTokens && Common._isCsvNullToken(trimmedValue)) {
853-
return null;
854-
}
855-
856865
const columnName = typeof context.column === 'string' ? context.column : String(context.column);
857-
const fieldType = resolveFieldType(columnName)?.toLowerCase();
858-
if (!fieldType) {
859-
return normalizedRawValue;
860-
}
861-
862-
return Common._castCsvValueByFieldType(normalizedRawValue, trimmedValue, fieldType);
866+
return Common._normalizeCsvCellValue(
867+
value,
868+
columnName,
869+
columnToColumnDataTypeMap,
870+
normalizedColumnDataTypeMap,
871+
preserveRawNullTokens
872+
);
863873
}
864874

865875
function resolveColumns(header: string[]): Array<string | undefined> {
866876
void columnToColumnDataTypeMap;
867877
return header;
868878
}
869879

870-
function resolveFieldType(columnName: string): string | undefined {
871-
const directFieldType = columnToColumnDataTypeMap?.get(columnName);
872-
if (directFieldType) {
873-
return directFieldType;
874-
}
875-
return normalizedColumnDataTypeMap.get(columnName.trim().toLowerCase());
876-
}
877-
878880
return new Promise<Array<Record<string, unknown>>>((resolve, reject) => {
879881
if (!fs.existsSync(filePath)) {
880882
resolve([]);
@@ -2074,6 +2076,89 @@ export class Common {
20742076
return this._CSV_NULL_TOKENS.has(value.trim().toLowerCase());
20752077
}
20762078

2079+
/**
2080+
* Normalizes column type map keys for case-insensitive lookups.
2081+
*
2082+
* @param columnToColumnDataTypeMap - Source column-to-type map.
2083+
* @returns Normalized type map.
2084+
*/
2085+
private static _normalizeCsvColumnDataTypeMap(columnToColumnDataTypeMap?: Map<string, string>): Map<string, string> {
2086+
const normalizedColumnDataTypeMap = new Map<string, string>();
2087+
columnToColumnDataTypeMap?.forEach((fieldType, columnName) => {
2088+
const normalizedName = columnName.trim().toLowerCase();
2089+
if (!normalizedName) {
2090+
return;
2091+
}
2092+
if (!normalizedColumnDataTypeMap.has(normalizedName)) {
2093+
normalizedColumnDataTypeMap.set(normalizedName, fieldType);
2094+
}
2095+
});
2096+
return normalizedColumnDataTypeMap;
2097+
}
2098+
2099+
/**
2100+
* Normalizes a CSV-like cell value using null-token and type casting rules.
2101+
*
2102+
* @param value - Raw value.
2103+
* @param columnName - Column name.
2104+
* @param columnToColumnDataTypeMap - Optional direct column type map.
2105+
* @param normalizedColumnDataTypeMap - Normalized column type map.
2106+
* @param preserveRawNullTokens - Keeps null tokens as raw string values when true.
2107+
* @returns Normalized value.
2108+
*/
2109+
private static _normalizeCsvCellValue(
2110+
value: unknown,
2111+
columnName: string,
2112+
columnToColumnDataTypeMap?: Map<string, string>,
2113+
normalizedColumnDataTypeMap: Map<string, string> = new Map<string, string>(),
2114+
preserveRawNullTokens = false
2115+
): unknown {
2116+
const rawValue = typeof value === 'string' ? value : String(value ?? '');
2117+
const normalizedRawValue = rawValue.replace(/\r+$/g, '');
2118+
const trimmedValue = normalizedRawValue.trim();
2119+
if (!trimmedValue.length) {
2120+
if (preserveRawNullTokens) {
2121+
return normalizedRawValue;
2122+
}
2123+
return Common.csvInsertNulls ? null : '';
2124+
}
2125+
2126+
if (!preserveRawNullTokens && Common._isCsvNullToken(trimmedValue)) {
2127+
return null;
2128+
}
2129+
2130+
const fieldType = Common._resolveCsvColumnFieldType(
2131+
columnName,
2132+
columnToColumnDataTypeMap,
2133+
normalizedColumnDataTypeMap
2134+
)?.toLowerCase();
2135+
if (!fieldType) {
2136+
return normalizedRawValue;
2137+
}
2138+
2139+
return Common._castCsvValueByFieldType(normalizedRawValue, trimmedValue, fieldType);
2140+
}
2141+
2142+
/**
2143+
* Resolves a CSV column type by exact or normalized column name.
2144+
*
2145+
* @param columnName - Column name.
2146+
* @param columnToColumnDataTypeMap - Optional direct column type map.
2147+
* @param normalizedColumnDataTypeMap - Normalized column type map.
2148+
* @returns Field type when found.
2149+
*/
2150+
private static _resolveCsvColumnFieldType(
2151+
columnName: string,
2152+
columnToColumnDataTypeMap?: Map<string, string>,
2153+
normalizedColumnDataTypeMap: Map<string, string> = new Map<string, string>()
2154+
): string | undefined {
2155+
const directFieldType = columnToColumnDataTypeMap?.get(columnName);
2156+
if (directFieldType) {
2157+
return directFieldType;
2158+
}
2159+
return normalizedColumnDataTypeMap.get(columnName.trim().toLowerCase());
2160+
}
2161+
20772162
/**
20782163
* Casts a CSV cell value based on field metadata type.
20792164
*

‎src/modules/models/job/MigrationJobTask.ts‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -521,9 +521,11 @@ export default class MigrationJobTask implements ISFdmuRunCustomAddonTask {
521521
logger.verboseFile(`[diagnostic] delete query: ${deleteQuery}`);
522522

523523
const orgDataService = new OrgDataService(this.job.script);
524+
const csvColumnDataTypeMap = this._createSourceCsvColumnDataTypeMap();
524525
const records = await orgDataService.queryOrgAsync(deleteQuery, this.targetData.org, {
525526
useBulk: this.targetData.useBulkQueryApi,
526527
useQueryAll: this.scriptObject.queryAllTarget || this.scriptObject.hardDelete,
528+
csvColumnDataTypeMap: csvColumnDataTypeMap.size > 0 ? csvColumnDataTypeMap : undefined,
527529
});
528530

529531
if (records.length === 0) {
@@ -1706,9 +1708,11 @@ export default class MigrationJobTask implements ISFdmuRunCustomAddonTask {
17061708
contactTask.targetData.queryCount += queries.length;
17071709

17081710
const orgDataService = new OrgDataService(this.job.script);
1711+
const csvColumnDataTypeMap = contactTask._createSourceCsvColumnDataTypeMap();
17091712
const records = await contactTask._retrieveFilteredRecordsAsync(queries, targetOrg, orgDataService, {
17101713
useBulk: contactTask.targetData.useBulkQueryApi,
17111714
useQueryAll: contactTask.scriptObject.queryAllTarget,
1715+
csvColumnDataTypeMap: csvColumnDataTypeMap.size > 0 ? csvColumnDataTypeMap : undefined,
17121716
});
17131717

17141718
if (records.length === 0) {
@@ -4485,7 +4489,7 @@ export default class MigrationJobTask implements ISFdmuRunCustomAddonTask {
44854489
queries: string[],
44864490
org: NonNullable<TaskOrgData['org']>,
44874491
orgDataService: OrgDataService,
4488-
options: { useBulk?: boolean; useQueryAll?: boolean }
4492+
options: { useBulk?: boolean; useQueryAll?: boolean; csvColumnDataTypeMap?: Map<string, string> }
44894493
): Promise<Array<Record<string, unknown>>> {
44904494
let records: Array<Record<string, unknown>> = [];
44914495
await Common.serialExecAsync(
@@ -4544,12 +4548,14 @@ export default class MigrationJobTask implements ISFdmuRunCustomAddonTask {
45444548
}
45454549
this.sourceData.queryCount += queries.length;
45464550
let records: Array<Record<string, unknown>> = [];
4551+
const csvColumnDataTypeMap = this._createSourceCsvColumnDataTypeMap();
45474552
await Common.serialExecAsync(
45484553
queries.map((query) => async () => {
45494554
this._logQueryString(query);
45504555
const batch = await orgDataService.queryOrgAsync(query, sourceOrg, {
45514556
useBulk: this.sourceData.useBulkQueryApi,
45524557
useQueryAll: this.scriptObject.useQueryAll,
4558+
csvColumnDataTypeMap: csvColumnDataTypeMap.size > 0 ? csvColumnDataTypeMap : undefined,
45534559
});
45544560
records = records.concat(batch);
45554561
return undefined;
@@ -4673,9 +4679,11 @@ export default class MigrationJobTask implements ISFdmuRunCustomAddonTask {
46734679
this._logQueryStart('queryingAll', 'source', 'org', queryMode);
46744680
this._logQueryString(query);
46754681
this.sourceData.queryCount += 1;
4682+
const csvColumnDataTypeMap = this._createSourceCsvColumnDataTypeMap();
46764683
const records = await orgDataService.queryOrgAsync(query, sourceOrg, {
46774684
useBulk: this.sourceData.useBulkQueryApi,
46784685
useQueryAll: this.scriptObject.useQueryAll,
4686+
csvColumnDataTypeMap: csvColumnDataTypeMap.size > 0 ? csvColumnDataTypeMap : undefined,
46794687
});
46804688
hasRecords = true;
46814689
this._finalizeRetrievedRecords(records, this.sourceData, 'source', false);
@@ -4684,9 +4692,11 @@ export default class MigrationJobTask implements ISFdmuRunCustomAddonTask {
46844692
if (queries.length > 0) {
46854693
this._logQueryStart('queryingIn', 'source', 'org', queryMode);
46864694
this.sourceData.queryCount += queries.length;
4695+
const csvColumnDataTypeMap = this._createSourceCsvColumnDataTypeMap();
46874696
const records = await this._retrieveFilteredRecordsAsync(queries, sourceOrg, orgDataService, {
46884697
useBulk: this.sourceData.useBulkQueryApi,
46894698
useQueryAll: this.scriptObject.useQueryAll,
4699+
csvColumnDataTypeMap: csvColumnDataTypeMap.size > 0 ? csvColumnDataTypeMap : undefined,
46904700
});
46914701
hasRecords = true;
46924702
this._finalizeRetrievedRecords(records, this.sourceData, 'source', false);
@@ -4760,19 +4770,23 @@ export default class MigrationJobTask implements ISFdmuRunCustomAddonTask {
47604770
this._logQueryStart('queryingAll', 'target', 'org', queryMode);
47614771
this._logQueryString(query);
47624772
this.targetData.queryCount += 1;
4773+
const csvColumnDataTypeMap = this._createSourceCsvColumnDataTypeMap();
47634774
records = await orgDataService.queryOrgAsync(query, targetOrg, {
47644775
useBulk: this.targetData.useBulkQueryApi,
47654776
useQueryAll: this.scriptObject.queryAllTarget,
4777+
csvColumnDataTypeMap: csvColumnDataTypeMap.size > 0 ? csvColumnDataTypeMap : undefined,
47664778
});
47674779
hasRecords = true;
47684780
} else {
47694781
const queries = this._createFilteredQueries(queryMode, false, fieldsInQuery);
47704782
if (queries.length > 0) {
47714783
this._logQueryStart('queryingIn', 'target', 'org', queryMode);
47724784
this.targetData.queryCount += queries.length;
4785+
const csvColumnDataTypeMap = this._createSourceCsvColumnDataTypeMap();
47734786
records = await this._retrieveFilteredRecordsAsync(queries, targetOrg, orgDataService, {
47744787
useBulk: this.targetData.useBulkQueryApi,
47754788
useQueryAll: this.scriptObject.queryAllTarget,
4789+
csvColumnDataTypeMap: csvColumnDataTypeMap.size > 0 ? csvColumnDataTypeMap : undefined,
47764790
});
47774791
hasRecords = true;
47784792
}

‎src/modules/org/OrgDataService.ts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -243,9 +243,12 @@ export default class OrgDataService {
243243
this._script.logger?.info(useBulk ? 'usingBulkAPIQuery' : 'usingRestApi', sObjectName, apiVersion);
244244
this._logQueryApiVersionDiagnostics(sObjectName, apiVersion, useBulk);
245245
}
246-
const records = useBulk
246+
const queriedRecords = useBulk
247247
? await this._queryBulkAsync(connection, query)
248248
: await this._queryRestAsync(connection, query, useQueryAll);
249+
const records = useBulk
250+
? Common.normalizeCsvRecordValues(queriedRecords, options.csvColumnDataTypeMap)
251+
: queriedRecords;
249252
this._writeSourceRecordsCache(query, org, records);
250253
return records;
251254
}

‎test/modules/org/org-data-service.test.ts‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -329,6 +329,57 @@ describe('OrgDataService', () => {
329329
}
330330
});
331331

332+
it('normalizes empty bulk query values using column type metadata', async () => {
333+
const script = new Script();
334+
335+
const org = new ScriptOrg();
336+
org.name = 'source';
337+
org.media = DATA_MEDIA_TYPE.Org;
338+
org.script = script;
339+
340+
const connection = {
341+
query: async (query: string) => {
342+
void query;
343+
throw new Error('REST query should not be used when bulk is requested.');
344+
},
345+
bulk: {
346+
pollInterval: 0,
347+
pollTimeout: 0,
348+
query: async (query: string) => {
349+
void query;
350+
const stream = new EventEmitter();
351+
setImmediate(() => {
352+
stream.emit('record', { Id: '003', LastName: 'Smith', Birthdate: '' });
353+
stream.emit('end');
354+
});
355+
return stream;
356+
},
357+
},
358+
};
359+
360+
const originalConnection = OrgConnectionAdapter.getConnectionForAliasAsync.bind(OrgConnectionAdapter);
361+
OrgConnectionAdapter.getConnectionForAliasAsync = async () => connection as never;
362+
363+
try {
364+
const service = new OrgDataService(script);
365+
const records = await service.queryOrgAsync('SELECT Id, LastName, Birthdate FROM Contact', org, {
366+
useBulk: true,
367+
csvColumnDataTypeMap: new Map<string, string>([
368+
['Id', 'id'],
369+
['LastName', 'string'],
370+
['Birthdate', 'date'],
371+
]),
372+
});
373+
374+
assert.equal(records.length, 1);
375+
assert.equal(records[0].Id, '003');
376+
assert.equal(records[0].LastName, 'Smith');
377+
assert.equal(records[0].Birthdate, null);
378+
} finally {
379+
OrgConnectionAdapter.getConnectionForAliasAsync = originalConnection;
380+
}
381+
});
382+
332383
it('returns cached source records without querying the org', async () => {
333384
const tempDir = createTempDir();
334385
const script = new Script();

0 commit comments

Comments
 (0)