tibrewalpratik17 commented on code in PR #13103:
URL: https://github.com/apache/pinot/pull/13103#discussion_r1622005593
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/SanitizationTransformer.java:
##########
@@ -32,51 +34,197 @@
* <ul>
* <li>No {@code null} characters in string values</li>
* <li>String values are within the length limit</li>
- * TODO: add length limit to BYTES values if necessary
* </ul>
* <p>NOTE: should put this after the {@link DataTypeTransformer} so that all
values follow the data types in
* {@link FieldSpec}.
+ * This uses the MaxLengthExceedStrategy in the {@link FieldSpec} to decide
what to do when the value exceeds the max.
+ * For TRIM_LENGTH, the value is trimmed to the max length.
+ * For SUBSTITUTE_DEFAULT_VALUE, the value is replaced with the default null
value string.
+ * For ERROR, an exception is thrown and the record is skipped.
+ * For NO_ACTION, the value is kept as is if no NULL_CHARACTER present else
trimmed till NULL.
+ * In the first 2 scenarios, this metric INCOMPLETE_REALTIME_ROWS_CONSUMED can
be tracked to know if a trimmed /
+ * default record was persisted.
+ * In the third scenario, this metric ROWS_WITH_ERRORS can be tracked to know
if a record was skipped.
+ * In the last scenario, this metric INCOMPLETE_REALTIME_ROWS_CONSUMED can be
tracked to know if a record was trimmed
+ * due to having a null character.
*/
public class SanitizationTransformer implements RecordTransformer {
- private final Map<String, Integer> _stringColumnMaxLengthMap = new
HashMap<>();
+ private static final String NULL_CHARACTER = "\0";
+ private final Map<String, SanitizedColumnInfo> _columnToColumnInfoMap = new
HashMap<>();
public SanitizationTransformer(Schema schema) {
+ FieldSpec.MaxLengthExceedStrategy maxLengthExceedStrategy;
for (FieldSpec fieldSpec : schema.getAllFieldSpecs()) {
- if (!fieldSpec.isVirtualColumn() && fieldSpec.getDataType() ==
DataType.STRING) {
- _stringColumnMaxLengthMap.put(fieldSpec.getName(),
fieldSpec.getMaxLength());
+ if (!fieldSpec.isVirtualColumn()) {
+ switch (fieldSpec.getDataType()) {
+ case STRING:
+ maxLengthExceedStrategy = fieldSpec.getMaxLengthExceedStrategy()
== null
+ ? FieldSpec.MaxLengthExceedStrategy.TRIM_LENGTH :
fieldSpec.getMaxLengthExceedStrategy();
+ _columnToColumnInfoMap.put(fieldSpec.getName(), new
SanitizedColumnInfo(fieldSpec.getName(),
+ fieldSpec.getMaxLength(), maxLengthExceedStrategy,
fieldSpec.getDefaultNullValueString()));
+ break;
+ case JSON:
+ case BYTES:
+ maxLengthExceedStrategy = fieldSpec.getMaxLengthExceedStrategy()
== null
+ ? FieldSpec.MaxLengthExceedStrategy.NO_ACTION :
fieldSpec.getMaxLengthExceedStrategy();
+ _columnToColumnInfoMap.put(fieldSpec.getName(), new
SanitizedColumnInfo(fieldSpec.getName(),
+ fieldSpec.getMaxLength(), maxLengthExceedStrategy,
fieldSpec.getDefaultNullValueString()));
+ break;
+ default:
+ // Do nothing for other data types
+ }
}
}
}
@Override
public boolean isNoOp() {
- return _stringColumnMaxLengthMap.isEmpty();
+ return _columnToColumnInfoMap.isEmpty();
}
@Override
public GenericRow transform(GenericRow record) {
- for (Map.Entry<String, Integer> entry :
_stringColumnMaxLengthMap.entrySet()) {
- String stringColumn = entry.getKey();
- int maxLength = entry.getValue();
- Object value = record.getValue(stringColumn);
- if (value instanceof String) {
- // Single-valued column
- String stringValue = (String) value;
- String sanitizedValue = StringUtil.sanitizeStringValue(stringValue,
maxLength);
- // NOTE: reference comparison
- //noinspection StringEquality
- if (sanitizedValue != stringValue) {
- record.putValue(stringColumn, sanitizedValue);
+ for (Map.Entry<String, SanitizedColumnInfo> entry :
_columnToColumnInfoMap.entrySet()) {
+ String columnName = entry.getKey();
+ Object value = record.getValue(columnName);
+ Pair<?, Boolean> result;
+ if (value instanceof byte[]) {
+ // Single-values BYTES column
+ result = sanitizeBytesValue(columnName, (byte[]) value,
entry.getValue());
+ record.putValue(columnName, result.getLeft());
+ if (result.getRight()) {
+ record.putValue(GenericRow.INCOMPLETE_RECORD_KEY, true);
Review Comment:
It was similar effort to remove this so i added `SANITIZED_RECORD_KEY`
instead
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
