[GitHub] [pinot] chenboat commented on a diff in pull request #9942: [feature] Add pinot-clp-log plugin to encode user-specified fields with CLP during ingestion

[GitBox <gi...@apache.org>]

chenboat commented on code in PR #9942:
URL: https://github.com/apache/pinot/pull/9942#discussion_r1044877771


##########
pinot-distribution/pinot-assembly.xml:
##########
@@ -146,6 +146,12 @@
       </source>
       <destName>plugins/pinot-input-format/pinot-avro/pinot-avro-${project.version}-shaded.jar</destName>
     </file>
+    <file>
+      <source>
+        ${pinot.root}/pinot-plugins/pinot-input-format/pinot-clp-log/target/pinot-clp-log-${project.version}-shaded.jar

Review Comment:
   why shaded jar is used? Are you seeing version conflict?



##########
pinot-plugins/pinot-input-format/pinot-clp-log/src/test/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractorTest.java:
##########
@@ -0,0 +1,110 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.plugin.inputformat.clplog;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.yscope.clp.compressorfrontend.MessageDecoder;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.fail;
+
+
+public class CLPLogRecordExtractorTest {

Review Comment:
   why is this an extractor test when it tests and decoder actually? Can we test the extractor directly?



##########
pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractor.java:
##########
@@ -0,0 +1,157 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.plugin.inputformat.clplog;
+
+import com.yscope.clp.compressorfrontend.EncodedMessage;
+import com.yscope.clp.compressorfrontend.MessageEncoder;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.data.readers.BaseRecordExtractor;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.data.readers.RecordExtractorConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A record extractor for log events. For configuration options, see {@link CLPLogRecordExtractorConfig}. This is an
+ * experimental feature.
+ * <p></p>
+ * The goal of this record extractor is to allow us to encode the fields specified in
+ * {@link CLPLogRecordExtractorConfig} using CLP. CLP is a compressor designed to encode unstructured log messages in a
+ * way that makes them more compressible. It does this by decomposing a message into three fields:

Review Comment:
   Make the comments complete by stating what to do with the fields NOT in the config.



##########
pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractor.java:
##########
@@ -0,0 +1,157 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.plugin.inputformat.clplog;
+
+import com.yscope.clp.compressorfrontend.EncodedMessage;
+import com.yscope.clp.compressorfrontend.MessageEncoder;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.data.readers.BaseRecordExtractor;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.data.readers.RecordExtractorConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A record extractor for log events. For configuration options, see {@link CLPLogRecordExtractorConfig}. This is an
+ * experimental feature.
+ * <p></p>
+ * The goal of this record extractor is to allow us to encode the fields specified in
+ * {@link CLPLogRecordExtractorConfig} using CLP. CLP is a compressor designed to encode unstructured log messages in a
+ * way that makes them more compressible. It does this by decomposing a message into three fields:
+ * <ul>
+ *   <li>the message's static text, called a log type;</li>
+ *   <li>repetitive variable values, called dictionary variables; and</li>
+ *   <li>non-repetitive variable values (called encoded variables since we encode them specially if possible).</li>
+ * </ul>
+ * For instance, if the field "message" is encoded, then the extractor will output:
+ * <ul>
+ *   <li>message_logtype</li>
+ *   <li>message_dictionaryVars</li>
+ *   <li>message_encodedVars</li>
+ * </ul>
+ * This class' implementation is based on {@link org.apache.pinot.plugin.inputformat.json.JSONRecordExtractor}.
+ */
+public class CLPLogRecordExtractor extends BaseRecordExtractor<Map<String, Object>> {
+  private static final Logger LOGGER = LoggerFactory.getLogger(CLPLogRecordExtractor.class);
+
+  private static final String LOGTYPE_COLUMN_SUFFIX = "_logtype";
+  private static final String DICTIONARY_VARS_COLUMN_SUFFIX = "_dictionaryVars";
+  private static final String ENCODED_VARS_COLUMN_SUFFIX = "_encodedVars";
+
+  private Set<String> _fields;
+  private boolean _extractAll = false;

Review Comment:
   add comments. What does extractAll mean?



##########
pinot-plugins/pinot-input-format/pinot-clp-log/src/test/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractorTest.java:
##########
@@ -0,0 +1,110 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.plugin.inputformat.clplog;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.yscope.clp.compressorfrontend.MessageDecoder;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.fail;
+
+
+public class CLPLogRecordExtractorTest {
+  @Test
+  void testCLPEncoding() {
+    // Setup decoder
+    CLPLogMessageDecoder messageDecoder = new CLPLogMessageDecoder();
+    Map<String, String> props = new HashMap<>();
+    Set<String> fieldsToRead = new HashSet<>();
+    // Add two fields for CLP encoding
+    props.put("fieldsForClpEncoding", "message1,message2");
+    fieldsToRead.add("message1_logtype");
+    fieldsToRead.add("message1_encodedVars");
+    fieldsToRead.add("message1_dictionaryVars");
+    fieldsToRead.add("message2_logtype");
+    fieldsToRead.add("message2_encodedVars");
+    fieldsToRead.add("message2_dictionaryVars");
+    // Add an unencoded field
+    fieldsToRead.add("timestamp");
+    try {
+      messageDecoder.init(props, fieldsToRead, null);

Review Comment:
   We also need to test different scenarios based on props's value.



##########
pinot-plugins/pinot-input-format/pinot-clp-log/src/main/java/org/apache/pinot/plugin/inputformat/clplog/CLPLogRecordExtractor.java:
##########
@@ -0,0 +1,157 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.plugin.inputformat.clplog;
+
+import com.yscope.clp.compressorfrontend.EncodedMessage;
+import com.yscope.clp.compressorfrontend.MessageEncoder;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.data.readers.BaseRecordExtractor;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.data.readers.RecordExtractorConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A record extractor for log events. For configuration options, see {@link CLPLogRecordExtractorConfig}. This is an
+ * experimental feature.
+ * <p></p>
+ * The goal of this record extractor is to allow us to encode the fields specified in
+ * {@link CLPLogRecordExtractorConfig} using CLP. CLP is a compressor designed to encode unstructured log messages in a
+ * way that makes them more compressible. It does this by decomposing a message into three fields:
+ * <ul>
+ *   <li>the message's static text, called a log type;</li>
+ *   <li>repetitive variable values, called dictionary variables; and</li>
+ *   <li>non-repetitive variable values (called encoded variables since we encode them specially if possible).</li>
+ * </ul>
+ * For instance, if the field "message" is encoded, then the extractor will output:
+ * <ul>
+ *   <li>message_logtype</li>
+ *   <li>message_dictionaryVars</li>
+ *   <li>message_encodedVars</li>
+ * </ul>
+ * This class' implementation is based on {@link org.apache.pinot.plugin.inputformat.json.JSONRecordExtractor}.
+ */
+public class CLPLogRecordExtractor extends BaseRecordExtractor<Map<String, Object>> {
+  private static final Logger LOGGER = LoggerFactory.getLogger(CLPLogRecordExtractor.class);
+
+  private static final String LOGTYPE_COLUMN_SUFFIX = "_logtype";
+  private static final String DICTIONARY_VARS_COLUMN_SUFFIX = "_dictionaryVars";
+  private static final String ENCODED_VARS_COLUMN_SUFFIX = "_encodedVars";
+
+  private Set<String> _fields;
+  private boolean _extractAll = false;
+  private CLPLogRecordExtractorConfig _config;
+
+  private EncodedMessage _clpEncodedMessage;
+  private MessageEncoder _clpMessageEncoder;
+
+  @Override
+  public void init(Set<String> fields, @Nullable RecordExtractorConfig recordExtractorConfig) {
+    _config = (CLPLogRecordExtractorConfig) recordExtractorConfig;
+    if (fields == null || fields.isEmpty()) {
+      _extractAll = true;
+      _fields = Collections.emptySet();
+    } else {
+      _fields = new HashSet<>(fields);
+      // Remove the fields to be CLP-encoded to make it easier to work with them
+      // in `extract`
+      _fields.removeAll(_config.getFieldsForClpEncoding());
+    }
+
+    _clpEncodedMessage = new EncodedMessage();
+    _clpMessageEncoder = new MessageEncoder();
+  }
+
+  @Override
+  public GenericRow extract(Map<String, Object> from, GenericRow to) {
+    Set<String> clpEncodedFieldNames = _config.getFieldsForClpEncoding();
+
+    if (_extractAll) {
+      for (Map.Entry<String, Object> recordEntry : from.entrySet()) {
+        String recordKey = recordEntry.getKey();
+        Object recordValue = recordEntry.getValue();
+        if (clpEncodedFieldNames.contains(recordKey)) {
+          encodeFieldWithClp(recordKey, recordValue, to);
+        } else {
+          if (null != recordValue) {
+            recordValue = convert(recordValue);
+          }
+          to.putValue(recordKey, recordValue);
+        }
+      }

Review Comment:
   Directly "return to;" and remove the else block.



-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org