This is an automated email from the ASF dual-hosted git repository.
lqc pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new da897dddaf Implement the ability to toggle dropping CLP processed
field from the original generic record. (#14534)
da897dddaf is described below
commit da897dddaf9d184fbc1051177615208b2aec8f97
Author: Jack Luo <[email protected]>
AuthorDate: Thu Nov 28 05:23:59 2024 +0800
Implement the ability to toggle dropping CLP processed field from the
original generic record. (#14534)
* Implement the ability to toggle dropping CLP processed field from the
original generic record.
* Fix unit test
---
.../inputformat/clplog/CLPLogRecordExtractor.java | 4 +++
.../clplog/CLPLogRecordExtractorConfig.java | 9 ++++++
.../clplog/CLPLogRecordExtractorTest.java | 32 +++++++++++++++-------
3 files changed, 35 insertions(+), 10 deletions(-)
diff --git
a/pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractor.java
b/pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractor.java
index f785a5bcce..eb52df9971 100644
---
a/pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractor.java
+++
b/pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractor.java
@@ -253,5 +253,9 @@ public class CLPLogRecordExtractor extends
BaseRecordExtractor<Map<String, Objec
to.putValue(key + ClpRewriter.LOGTYPE_COLUMN_SUFFIX, logtype);
to.putValue(key + ClpRewriter.DICTIONARY_VARS_COLUMN_SUFFIX, dictVars);
to.putValue(key + ClpRewriter.ENCODED_VARS_COLUMN_SUFFIX, encodedVars);
+
+ if (!_config.getRemoveProcessedFields()) {
+ to.putValue(key, value);
+ }
}
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractorConfig.java
b/pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractorConfig.java
index 87d5b8efe6..3d08f42da5 100644
---
a/pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractorConfig.java
+++
b/pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractorConfig.java
@@ -46,6 +46,7 @@ import org.slf4j.LoggerFactory;
public class CLPLogRecordExtractorConfig implements RecordExtractorConfig {
public static final String FIELDS_FOR_CLP_ENCODING_CONFIG_KEY =
"fieldsForClpEncoding";
public static final String FIELDS_FOR_CLP_ENCODING_SEPARATOR = ",";
+ public static final String REMOVE_PROCESSED_FIELDS_CONFIG_KEY =
"removeProcessedFields";
public static final String UNENCODABLE_FIELD_SUFFIX_CONFIG_KEY =
"unencodableFieldSuffix";
public static final String UNENCODABLE_FIELD_ERROR_CONFIG_KEY =
"unencodableFieldError";
@@ -54,6 +55,7 @@ public class CLPLogRecordExtractorConfig implements
RecordExtractorConfig {
private final Set<String> _fieldsForClpEncoding = new HashSet<>();
private String _unencodableFieldSuffix = null;
private String _unencodableFieldError = null;
+ private boolean _removeProcessedFields = false;
@Override
public void init(Map<String, String> props) {
@@ -75,6 +77,9 @@ public class CLPLogRecordExtractorConfig implements
RecordExtractorConfig {
}
}
+ // True if field is specified and equal to true (ignore case), false for
all other cases
+ _removeProcessedFields =
Boolean.parseBoolean(props.get(REMOVE_PROCESSED_FIELDS_CONFIG_KEY));
+
String unencodableFieldSuffix =
props.get(UNENCODABLE_FIELD_SUFFIX_CONFIG_KEY);
if (null != unencodableFieldSuffix) {
if (unencodableFieldSuffix.length() == 0) {
@@ -98,6 +103,10 @@ public class CLPLogRecordExtractorConfig implements
RecordExtractorConfig {
return _fieldsForClpEncoding;
}
+ public boolean getRemoveProcessedFields() {
+ return _removeProcessedFields;
+ }
+
public String getUnencodableFieldSuffix() {
return _unencodableFieldSuffix;
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-clp-log/src/test/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractorTest.java
b/pinot-plugins/pinot-input-format/pinot-clp-log/src/test/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractorTest.java
index 0ad198fe18..2e002cdb0c 100644
---
a/pinot-plugins/pinot-input-format/pinot-clp-log/src/test/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractorTest.java
+++
b/pinot-plugins/pinot-input-format/pinot-clp-log/src/test/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractorTest.java
@@ -32,6 +32,7 @@ import org.testng.annotations.Test;
import static
org.apache.pinot.plugin.inputformat.clplog.CLPLogRecordExtractorConfig.FIELDS_FOR_CLP_ENCODING_CONFIG_KEY;
import static
org.apache.pinot.plugin.inputformat.clplog.CLPLogRecordExtractorConfig.FIELDS_FOR_CLP_ENCODING_SEPARATOR;
+import static
org.apache.pinot.plugin.inputformat.clplog.CLPLogRecordExtractorConfig.REMOVE_PROCESSED_FIELDS_CONFIG_KEY;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNull;
@@ -54,11 +55,17 @@ public class CLPLogRecordExtractorTest {
@Test
public void testCLPEncoding() {
+ testCLPEncoding(true);
+ testCLPEncoding(false);
+ }
+
+ public void testCLPEncoding(boolean removeProcessedField) {
Map<String, String> props = new HashMap<>();
Set<String> fieldsToRead = new HashSet<>();
// Add some fields for CLP encoding
props.put(FIELDS_FOR_CLP_ENCODING_CONFIG_KEY,
_MESSAGE_1_FIELD_NAME + FIELDS_FOR_CLP_ENCODING_SEPARATOR +
_MESSAGE_2_FIELD_NAME);
+ props.put(REMOVE_PROCESSED_FIELDS_CONFIG_KEY,
String.valueOf(removeProcessedField));
addCLPEncodedField(_MESSAGE_1_FIELD_NAME, fieldsToRead);
addCLPEncodedField(_MESSAGE_2_FIELD_NAME, fieldsToRead);
// Add some unencoded fields
@@ -70,25 +77,27 @@ public class CLPLogRecordExtractorTest {
row = extract(props, fieldsToRead);
assertEquals(row.getValue(_TIMESTAMP_FIELD_NAME), _TIMESTAMP_FIELD_VALUE);
assertNull(row.getValue(_LEVEL_FIELD_NAME));
- validateClpEncodedField(row, _MESSAGE_1_FIELD_NAME,
_MESSAGE_1_FIELD_VALUE);
- validateClpEncodedField(row, _MESSAGE_2_FIELD_NAME,
_MESSAGE_2_FIELD_VALUE);
+ validateClpEncodedField(row, _MESSAGE_1_FIELD_NAME,
_MESSAGE_1_FIELD_VALUE, removeProcessedField);
+ validateClpEncodedField(row, _MESSAGE_2_FIELD_NAME,
_MESSAGE_2_FIELD_VALUE, removeProcessedField);
// Test extracting all fields
row = extract(props, null);
assertEquals(row.getValue(_TIMESTAMP_FIELD_NAME), _TIMESTAMP_FIELD_VALUE);
assertEquals(row.getValue(_LEVEL_FIELD_NAME), _LEVEL_FIELD_VALUE);
- validateClpEncodedField(row, _MESSAGE_1_FIELD_NAME,
_MESSAGE_1_FIELD_VALUE);
- validateClpEncodedField(row, _MESSAGE_2_FIELD_NAME,
_MESSAGE_2_FIELD_VALUE);
+ validateClpEncodedField(row, _MESSAGE_1_FIELD_NAME,
_MESSAGE_1_FIELD_VALUE, removeProcessedField);
+ validateClpEncodedField(row, _MESSAGE_2_FIELD_NAME,
_MESSAGE_2_FIELD_VALUE, removeProcessedField);
}
@Test
public void testBadCLPEncodingConfig() {
Map<String, String> props = new HashMap<>();
Set<String> fieldsToRead = new HashSet<>();
+ boolean removeProcessedField = true;
// Add some fields for CLP encoding with some mistakenly empty field names
String separator = FIELDS_FOR_CLP_ENCODING_SEPARATOR;
props.put(FIELDS_FOR_CLP_ENCODING_CONFIG_KEY, separator +
_MESSAGE_1_FIELD_NAME
+ separator + separator + _MESSAGE_2_FIELD_NAME + separator);
+ props.put(REMOVE_PROCESSED_FIELDS_CONFIG_KEY,
String.valueOf(removeProcessedField));
addCLPEncodedField(_MESSAGE_1_FIELD_NAME, fieldsToRead);
addCLPEncodedField(_MESSAGE_2_FIELD_NAME, fieldsToRead);
// Add some unencoded fields
@@ -100,16 +109,16 @@ public class CLPLogRecordExtractorTest {
row = extract(props, fieldsToRead);
assertEquals(row.getValue(_TIMESTAMP_FIELD_NAME), _TIMESTAMP_FIELD_VALUE);
assertNull(row.getValue(_LEVEL_FIELD_NAME));
- validateClpEncodedField(row, _MESSAGE_1_FIELD_NAME,
_MESSAGE_1_FIELD_VALUE);
- validateClpEncodedField(row, _MESSAGE_2_FIELD_NAME,
_MESSAGE_2_FIELD_VALUE);
+ validateClpEncodedField(row, _MESSAGE_1_FIELD_NAME,
_MESSAGE_1_FIELD_VALUE, removeProcessedField);
+ validateClpEncodedField(row, _MESSAGE_2_FIELD_NAME,
_MESSAGE_2_FIELD_VALUE, removeProcessedField);
// Test extracting all fields
row = extract(props, null);
assertEquals(row.getValue(_TIMESTAMP_FIELD_NAME), _TIMESTAMP_FIELD_VALUE);
assertEquals(row.getValue(_LEVEL_FIELD_NAME), _LEVEL_FIELD_VALUE);
assertEquals(row.getValue(_BOOLEAN_FIELD_NAME), _BOOLEAN_FIELD_VALUE);
- validateClpEncodedField(row, _MESSAGE_1_FIELD_NAME,
_MESSAGE_1_FIELD_VALUE);
- validateClpEncodedField(row, _MESSAGE_2_FIELD_NAME,
_MESSAGE_2_FIELD_VALUE);
+ validateClpEncodedField(row, _MESSAGE_1_FIELD_NAME,
_MESSAGE_1_FIELD_VALUE, removeProcessedField);
+ validateClpEncodedField(row, _MESSAGE_2_FIELD_NAME,
_MESSAGE_2_FIELD_VALUE, removeProcessedField);
}
@Test
@@ -163,10 +172,13 @@ public class CLPLogRecordExtractorTest {
return row;
}
- private void validateClpEncodedField(GenericRow row, String fieldName,
String expectedFieldValue) {
+ private void validateClpEncodedField(GenericRow row, String fieldName,
String expectedFieldValue,
+ boolean removeProcessedField) {
try {
// Decode and validate field
- assertNull(row.getValue(fieldName));
+ if (removeProcessedField) {
+ assertNull(row.getValue(fieldName));
+ }
String logtype = (String) row.getValue(fieldName +
ClpRewriter.LOGTYPE_COLUMN_SUFFIX);
assertNotEquals(logtype, null);
String[] dictionaryVars =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
