Skip to content

Commit 8ef7b56

Browse files
inv-jishnuinv-jishnu
committed
Resolve minor bugs in data loader core (#2752)
1 parent b7aac6d commit 8ef7b56

File tree

5 files changed

+51
-31
lines changed

5 files changed

+51
-31
lines changed

‎core/src/main/java/com/scalar/db/common/error/CoreError.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -689,13 +689,13 @@ public enum CoreError implements ScalarDbError {
689689
DATA_LOADER_INVALID_BASE64_ENCODING_FOR_COLUMN_VALUE(
690690
Category.USER_ERROR,
691691
"0149",
692-
"Invalid base64 encoding for blob value for column %s in table %s in namespace %s",
692+
"Invalid base64 encoding for blob value '%s' for column %s in table %s in namespace %s",
693693
"",
694694
""),
695695
DATA_LOADER_INVALID_NUMBER_FORMAT_FOR_COLUMN_VALUE(
696696
Category.USER_ERROR,
697697
"0150",
698-
"Invalid number specified for column %s in table %s in namespace %s",
698+
"Invalid number '%s' specified for column %s in table %s in namespace %s",
699699
"",
700700
""),
701701
DATA_LOADER_ERROR_METHOD_NULL_ARGUMENT(
@@ -898,7 +898,7 @@ public enum CoreError implements ScalarDbError {
898898
DATA_LOADER_INVALID_DATE_TIME_FOR_COLUMN_VALUE(
899899
Category.USER_ERROR,
900900
"0199",
901-
"Invalid date time value specified for column %s in table %s in namespace %s.",
901+
"Invalid date time value '%s' specified for column %s in table %s in namespace %s.",
902902
"",
903903
""),
904904
DATA_LOADER_NULL_OR_EMPTY_KEY_VALUE_INPUT(

‎data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/processor/ImportProcessor.java‎

Lines changed: 5 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -416,19 +416,11 @@ private ImportDataChunkStatus processDataChunkWithTransactions(
416416
ImportTransactionBatchResult importTransactionBatchResult =
417417
processTransactionBatch(dataChunk.getDataChunkId(), transactionBatch);
418418

419-
importTransactionBatchResult
420-
.getRecords()
421-
.forEach(
422-
batchRecords -> {
423-
if (batchRecords.getTargets().stream()
424-
.allMatch(
425-
targetResult ->
426-
targetResult.getStatus().equals(ImportTargetResultStatus.SAVED))) {
427-
successCount.incrementAndGet();
428-
} else {
429-
failureCount.incrementAndGet();
430-
}
431-
});
419+
if (importTransactionBatchResult.isSuccess()) {
420+
successCount.addAndGet(importTransactionBatchResult.getRecords().size());
421+
} else {
422+
failureCount.addAndGet(importTransactionBatchResult.getRecords().size());
423+
}
432424
}
433425
Instant endTime = Instant.now();
434426
int totalDuration = (int) Duration.between(startTime, endTime).toMillis();

‎data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/task/ImportTask.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ public ImportTaskResult execute() {
7373
return ImportTaskResult.builder()
7474
.rawRecord(params.getSourceRecord())
7575
.rowNumber(params.getRowNumber())
76-
.targets(Collections.singletonList(singleTargetResult))
76+
.targets(new ArrayList<>(Collections.singletonList(singleTargetResult)))
7777
.build();
7878
}
7979

@@ -308,11 +308,14 @@ && shouldRevalidateMissingColumns(importOptions, checkForMissingColumns)) {
308308
optionalScalarDBResult.orElse(null),
309309
mutableSourceRecord,
310310
importOptions.isIgnoreNullValues(),
311-
tableMetadata);
311+
tableMetadata,
312+
namespace,
313+
table);
312314
} catch (Base64Exception | ColumnParsingException e) {
313315
return ImportTargetResult.builder()
314316
.namespace(namespace)
315317
.tableName(table)
318+
.importedRecord(mutableSourceRecord)
316319
.status(ImportTargetResultStatus.VALIDATION_FAILED)
317320
.errors(Collections.singletonList(e.getMessage()))
318321
.build();

‎data-loader/core/src/main/java/com/scalar/db/dataloader/core/util/ColumnUtils.java‎

Lines changed: 24 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -134,17 +134,17 @@ public static Column<?> createColumnFromValue(
134134
} catch (NumberFormatException e) {
135135
throw new ColumnParsingException(
136136
CoreError.DATA_LOADER_INVALID_NUMBER_FORMAT_FOR_COLUMN_VALUE.buildMessage(
137-
columnName, columnInfo.getTableName(), columnInfo.getNamespace()),
137+
value, columnName, columnInfo.getTableName(), columnInfo.getNamespace()),
138138
e);
139139
} catch (DateTimeParseException e) {
140140
throw new ColumnParsingException(
141141
CoreError.DATA_LOADER_INVALID_DATE_TIME_FOR_COLUMN_VALUE.buildMessage(
142-
columnName, columnInfo.getTableName(), columnInfo.getNamespace()),
142+
value, columnName, columnInfo.getTableName(), columnInfo.getNamespace()),
143143
e);
144144
} catch (IllegalArgumentException e) {
145145
throw new ColumnParsingException(
146146
CoreError.DATA_LOADER_INVALID_BASE64_ENCODING_FOR_COLUMN_VALUE.buildMessage(
147-
columnName, columnInfo.getTableName(), columnInfo.getNamespace()),
147+
value, columnName, columnInfo.getTableName(), columnInfo.getNamespace()),
148148
e);
149149
}
150150
}
@@ -166,6 +166,8 @@ public static Column<?> createColumnFromValue(
166166
* @param sourceRecord the source data in JSON format to compare against
167167
* @param ignoreNullValues if true, null values will be excluded from the result
168168
* @param tableMetadata metadata about the table structure and column types
169+
* @param namespace namespace in which the table is present
170+
* @param table table name to which data is to be imported
169171
* @return a List of Column objects representing the processed data
170172
* @throws Base64Exception if there's an error processing base64 encoded BLOB data
171173
* @throws ColumnParsingException if there's an error parsing column values
@@ -174,7 +176,9 @@ public static List<Column<?>> getColumnsFromResult(
174176
Result scalarDBResult,
175177
JsonNode sourceRecord,
176178
boolean ignoreNullValues,
177-
TableMetadata tableMetadata)
179+
TableMetadata tableMetadata,
180+
String namespace,
181+
String table)
178182
throws Base64Exception, ColumnParsingException {
179183

180184
List<Column<?>> columns = new ArrayList<>();
@@ -193,7 +197,9 @@ public static List<Column<?>> getColumnsFromResult(
193197
sourceRecord,
194198
columnName,
195199
ignoreNullValues,
196-
tableMetadata.getColumnDataTypes());
200+
tableMetadata.getColumnDataTypes(),
201+
namespace,
202+
table);
197203

198204
if (column != null) {
199205
columns.add(column);
@@ -242,6 +248,8 @@ private static Set<String> getColumnsToIgnore(
242248
* @param columnName the name of the column to retrieve
243249
* @param ignoreNullValues whether to ignore null values in the result
244250
* @param dataTypesByColumns mapping of column names to their data types
251+
* @param namespace namespace in which the table is present
252+
* @param table table name to which data is to be imported
245253
* @return the Column object containing the value, or null if ignored
246254
* @throws ColumnParsingException if there's an error parsing the column value
247255
*/
@@ -250,13 +258,15 @@ private static Column<?> getColumn(
250258
JsonNode sourceRecord,
251259
String columnName,
252260
boolean ignoreNullValues,
253-
Map<String, DataType> dataTypesByColumns)
261+
Map<String, DataType> dataTypesByColumns,
262+
String namespace,
263+
String table)
254264
throws ColumnParsingException {
255265
if (scalarDBResult != null && !sourceRecord.has(columnName)) {
256266
return getColumnFromResult(scalarDBResult, columnName);
257267
} else {
258268
return getColumnFromSourceRecord(
259-
sourceRecord, columnName, ignoreNullValues, dataTypesByColumns);
269+
sourceRecord, columnName, ignoreNullValues, dataTypesByColumns, namespace, table);
260270
}
261271
}
262272

@@ -279,22 +289,27 @@ private static Column<?> getColumnFromResult(Result scalarDBResult, String colum
279289
* @param columnName column name
280290
* @param ignoreNullValues ignore null values or not
281291
* @param dataTypesByColumns data types of columns
292+
* @param namespace namespace in which the table is present
293+
* @param table table name to which data is to be imported
282294
* @return column data
283295
* @throws ColumnParsingException if an error occurs while parsing the column
284296
*/
285297
private static Column<?> getColumnFromSourceRecord(
286298
JsonNode sourceRecord,
287299
String columnName,
288300
boolean ignoreNullValues,
289-
Map<String, DataType> dataTypesByColumns)
301+
Map<String, DataType> dataTypesByColumns,
302+
String namespace,
303+
String table)
290304
throws ColumnParsingException {
291305
DataType dataType = dataTypesByColumns.get(columnName);
292306
String columnValue =
293307
sourceRecord.has(columnName) && !sourceRecord.get(columnName).isNull()
294308
? sourceRecord.get(columnName).asText()
295309
: null;
296310
if (!ignoreNullValues || columnValue != null) {
297-
ColumnInfo columnInfo = ColumnInfo.builder().columnName(columnName).build();
311+
ColumnInfo columnInfo =
312+
ColumnInfo.builder().columnName(columnName).tableName(table).namespace(namespace).build();
298313
return createColumnFromValue(dataType, columnInfo, columnValue);
299314
}
300315
return null;

‎data-loader/core/src/test/java/com/scalar/db/dataloader/core/util/ColumnUtilsTest.java‎

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -161,7 +161,7 @@ void createColumnFromValue_invalidNumberFormat_throwsNumberFormatException() {
161161
() -> ColumnUtils.createColumnFromValue(DataType.INT, columnInfo, value));
162162
assertEquals(
163163
CoreError.DATA_LOADER_INVALID_NUMBER_FORMAT_FOR_COLUMN_VALUE.buildMessage(
164-
columnName, "table", "ns"),
164+
value, columnName, "table", "ns"),
165165
exception.getMessage());
166166
}
167167

@@ -181,7 +181,7 @@ void createColumnFromValue_invalidBase64_throwsBase64Exception() {
181181
() -> ColumnUtils.createColumnFromValue(DataType.BLOB, columnInfo, value));
182182
assertEquals(
183183
CoreError.DATA_LOADER_INVALID_BASE64_ENCODING_FOR_COLUMN_VALUE.buildMessage(
184-
columnName, "table", "ns"),
184+
value, columnName, "table", "ns"),
185185
exception.getMessage());
186186
}
187187
/**
@@ -200,7 +200,7 @@ void createColumnFromValue_invalidDateTimeFormat_throwsDateTimeParseException()
200200
() -> ColumnUtils.createColumnFromValue(DataType.TIMESTAMP, columnInfo, value));
201201
assertEquals(
202202
CoreError.DATA_LOADER_INVALID_DATE_TIME_FOR_COLUMN_VALUE.buildMessage(
203-
columnName, "table", "ns"),
203+
value, columnName, "table", "ns"),
204204
exception.getMessage());
205205
}
206206

@@ -215,7 +215,17 @@ void createColumnFromValue_invalidDateTimeFormat_throwsDateTimeParseException()
215215
void getColumnsFromResult_withValidData_shouldReturnColumns()
216216
throws Base64Exception, ColumnParsingException {
217217
List<Column<?>> columns =
218-
ColumnUtils.getColumnsFromResult(scalarDBResult, sourceRecord, false, mockMetadata);
218+
ColumnUtils.getColumnsFromResult(
219+
scalarDBResult, sourceRecord, false, mockMetadata, "namespace", "table");
220+
assertEquals(8, columns.size());
221+
}
222+
223+
@Test
224+
void getColumnsFromResult_withResultNull_withValidData_shouldReturnColumns()
225+
throws Base64Exception, ColumnParsingException {
226+
List<Column<?>> columns =
227+
ColumnUtils.getColumnsFromResult(
228+
null, sourceRecord, false, mockMetadata, "namespace", "table");
219229
assertEquals(8, columns.size());
220230
}
221231
}

0 commit comments

Comments
 (0)