You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "jackluo923 (via GitHub)" <gi...@apache.org> on 2024/10/15 18:08:22 UTC
[PR] Add the initial implementation of CLPMutableForwardIndexV2 [pinot]
jackluo923 opened a new pull request, #14241:
URL: https://github.com/apache/pinot/pull/14241
Placeholder
--
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
Re: [PR] Add the initial implementation of CLPMutableForwardIndexV2 [pinot]
Posted by "jackluo923 (via GitHub)" <gi...@apache.org>.
jackluo923 commented on code in PR #14241:
URL: https://github.com/apache/pinot/pull/14241#discussion_r1802041475
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/CLPMutableForwardIndexV2.java:
##########
@@ -0,0 +1,468 @@
+/**
+ * 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.segment.local.realtime.impl.forward;
+
+import com.yscope.clp.compressorfrontend.BuiltInVariableHandlingRuleVersions;
+import com.yscope.clp.compressorfrontend.EmptyArrayUtils;
+import com.yscope.clp.compressorfrontend.EncodedMessage;
+import com.yscope.clp.compressorfrontend.FlattenedByteArray;
+import com.yscope.clp.compressorfrontend.FlattenedByteArrayFactory;
+import com.yscope.clp.compressorfrontend.MessageDecoder;
+import com.yscope.clp.compressorfrontend.MessageEncoder;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import javax.validation.constraints.NotNull;
+import org.apache.pinot.segment.local.realtime.impl.dictionary.BytesOffHeapMutableDictionary;
+import org.apache.pinot.segment.spi.index.mutable.MutableForwardIndex;
+import org.apache.pinot.segment.spi.memory.PinotDataBufferMemoryManager;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This mutable forward index implements a composite index for string-typed columns with dynamic encoding options:
+ * <ol>
+ * <li>Pure CLP dictionary encoding when the dictionary cardinality is below a configurable threshold.</li>
+ * <li>CLP dictionary encoding combined with a raw string forward index when the dictionary cardinality exceeds
+ * the threshold.</li>
+ * </ol>
+ * <p>
+ * Initially, CLP encoding transforms a high-cardinality log message string into three data columns:
+ * <ul>
+ * <li>Logtype (very low cardinality) - essentially an inferred format string of the log</li>
+ * <li>Dictionary variables (medium cardinality) - variables with both alphabets and numbers</li>
+ * <li>Encoded variables (high cardinality) - pure fixed point and floating point numbers</li>
+ * </ul>
+ * The logtype and dictionary variables are dictionary-encoded, while the encoded variables are stored as longs.
+ * Notably, both {@code encodedVarIds} and {@code encodedVars} are multi-valued, but they are stored using a
+ * flattened single-value mutable forward index, along with a separate forward index to capture the end offsets
+ * for each multi-valued document. This approach is necessary because the maximum number of values per document
+ * is unknown during ingestion, unlike the existing multi-value forward index, which requires this information
+ * upfront. During the conversion from mutable to immutable forward index, the two single-value mutable indices
+ * are merged into a single immutable multi-valued forward index, as the max length is known at conversion time.
+ * <p>
+ * During ingestion, if the cardinality of either the {@code logtypeId} or {@code dictVarID} exceeds a predefined
+ * threshold, the ingestion mode switches to a raw bytes forward index for subsequent documents. Maintaining very
+ * large dictionaries is inefficient in Pinot due to memory and I/O constraints (memory-mapped). Switching to a
+ * raw bytes forward index helps avoid these issues. During reads, if the requested {@code docId} is in the raw
+ * forward index, the raw bytes are returned. Otherwise, the log type, dictionary variables, and encoded variables
+ * are decoded using the CLP decoder to return the original log message's bytes.
+ *
+ * <p><b>Note on Write and Read Operations:</b> Writes are strictly sequential, while reads can be performed
+ * randomly. The supported append operations are:</p>
+ * <ul>
+ * <li>{@code setString(int docId, String value)} - Encodes the log message using CLP and invokes
+ * {@code appendEncodedMessage(@NotNull EncodedMessage clpEncodedMessage)}.</li>
+ * <li>{@code appendEncodedMessage(@NotNull EncodedMessage clpEncodedMessage)}</li>
+ * </ul>
+ *
+ * <p><b>Limitations:</b> The current CLP mutable forward index does not achieve the same compression ratio as the
+ * original standalone CLP implementation, primarily due to design differences between Pinot and CLP. While Pinot
+ * is optimized for fast random access, CLP is designed for single-pass streaming compression and search. As a result,
+ * Pinot sacrifices some compression efficiency for these primary reasons:
+ * <ul>
+ * <li>Pinot implementation uses block compression compared to CLP’s standalone streaming compression, which achieves
+ * much better compression ratio at the cost of random access performance.</li>
+ * <li>Pinot implementation uses uncompressed dictionaries for random lookups of log types and dictionary variables,
+ * whereas CLP employs compressed dictionaries suited only for single-pass streaming queries.</li>
+ * <li>Pinot stores additional offsets and length metadata for log types, dictionary variables, and encoded
+ * variables:</li>
+ * <ul>
+ * <li>Streaming forward indices used by CLP's standalone implementation, do not need to store document start or
+ * end markers because the boundary of one document naturally aligns with the next.</li>
+ * <li>CLP avoids storing the number of {@code dictVars} and {@code encodedVars}, as this information is already
+ * embedded in the log type and available during decoding. Pinot, however, needs to store this metadata,
+ * which can sometimes take up more space than the data itself when compressed.</li>
+ * </ul>
+ * </ul>
+ * Additionally, CLP standalone binaries can perform search without decompressing the data into plain text. The most
+ * common query type on log data is partial matching. Frequently, searches can be completed by scanning only the logtype
+ * dictionaries for partial matches. Searches can also be executed on {@code logtypeId} and {@code dictVarId} directly
+ * by first performing lookup on the dictionaries to get a subset of dictionary ids to filter, whereas in Pinot,
+ * each dictionary id must be first converted back to strings, followed by a brute-force search on the corresponding
+ * string value. For these reasons, direct searching on CLP columns in Pinot is not yet implemented but may be
+ * included in future updates.</p>
+ */
+public class CLPMutableForwardIndexV2 implements MutableForwardIndex {
+ protected static final Logger LOGGER = LoggerFactory.getLogger(CLPMutableForwardIndexV2.class);
+ public final String _columnName;
+
+ protected final EncodedMessage _clpEncodedMessage;
+ protected final MessageEncoder _clpMessageEncoder;
+ protected final MessageDecoder _clpMessageDecoder;
+
+ protected int _nextDocId = 0;
+ protected int _nextDictVarDocId = 0;
+ protected int _nextEncodedVarId = 0;
+ protected int _bytesRawFwdIndexDocIdStartOffset = Integer.MAX_VALUE;
Review Comment:
Yes, this is only used for CLPV2 (but might be used for CLPV3 in the future). Internally, we have classes which extends the functionality of this class, so some class member variables are marked as protected.
--
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
Re: [PR] Add the initial implementation of CLPMutableForwardIndexV2 [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #14241:
URL: https://github.com/apache/pinot/pull/14241#issuecomment-2415262460
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/14241?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 0.00%. Comparing base [(`59551e4`)](https://app.codecov.io/gh/apache/pinot/commit/59551e45224f1535c4863fd577622b37366ccc97?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`370d907`)](https://app.codecov.io/gh/apache/pinot/commit/370d907140a91a88399f0bbdf69e0fa044454f00?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 1188 commits behind head on master.
> :exclamation: There is a different number of reports uploaded between BASE (59551e4) and HEAD (370d907). Click for more details.
>
> <details><summary>HEAD has 48 uploads less than BASE</summary>
>
>| Flag | BASE (59551e4) | HEAD (370d907) |
>|------|------|------|
>|integration|7|2|
>|integration2|3|2|
>|temurin|12|2|
>|java-21|7|2|
>|skip-bytebuffers-true|3|1|
>|skip-bytebuffers-false|7|1|
>|unittests|5|0|
>|unittests1|2|0|
>|java-11|5|0|
>|unittests2|3|0|
>|integration1|2|0|
>|custom-integration1|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #14241 +/- ##
=============================================
- Coverage 61.75% 0.00% -61.76%
=============================================
Files 2436 3 -2433
Lines 133233 6 -133227
Branches 20636 0 -20636
=============================================
- Hits 82274 0 -82274
+ Misses 44911 6 -44905
+ Partials 6048 0 -6048
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/14241/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/14241/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/14241/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-0.01%)` | :arrow_down: |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/14241/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/14241/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (ø)` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/14241/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/14241/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-61.63%)` | :arrow_down: |
| [skip-bytebuffers-false](https://app.codecov.io/gh/apache/pinot/pull/14241/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-61.75%)` | :arrow_down: |
| [skip-bytebuffers-true](https://app.codecov.io/gh/apache/pinot/pull/14241/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-27.73%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/14241/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-61.76%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/14241/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/14241/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/14241/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/14241?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
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
Re: [PR] Add the initial implementation of CLPMutableForwardIndexV2 [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14241:
URL: https://github.com/apache/pinot/pull/14241#discussion_r1801989306
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/CLPMutableForwardIndexV2.java:
##########
@@ -0,0 +1,468 @@
+/**
+ * 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.segment.local.realtime.impl.forward;
+
+import com.yscope.clp.compressorfrontend.BuiltInVariableHandlingRuleVersions;
+import com.yscope.clp.compressorfrontend.EmptyArrayUtils;
+import com.yscope.clp.compressorfrontend.EncodedMessage;
+import com.yscope.clp.compressorfrontend.FlattenedByteArray;
+import com.yscope.clp.compressorfrontend.FlattenedByteArrayFactory;
+import com.yscope.clp.compressorfrontend.MessageDecoder;
+import com.yscope.clp.compressorfrontend.MessageEncoder;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import javax.validation.constraints.NotNull;
+import org.apache.pinot.segment.local.realtime.impl.dictionary.BytesOffHeapMutableDictionary;
+import org.apache.pinot.segment.spi.index.mutable.MutableForwardIndex;
+import org.apache.pinot.segment.spi.memory.PinotDataBufferMemoryManager;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This mutable forward index implements a composite index for string-typed columns with dynamic encoding options:
+ * <ol>
+ * <li>Pure CLP dictionary encoding when the dictionary cardinality is below a configurable threshold.</li>
+ * <li>CLP dictionary encoding combined with a raw string forward index when the dictionary cardinality exceeds
+ * the threshold.</li>
+ * </ol>
+ * <p>
+ * Initially, CLP encoding transforms a high-cardinality log message string into three data columns:
+ * <ul>
+ * <li>Logtype (very low cardinality) - essentially an inferred format string of the log</li>
+ * <li>Dictionary variables (medium cardinality) - variables with both alphabets and numbers</li>
+ * <li>Encoded variables (high cardinality) - pure fixed point and floating point numbers</li>
+ * </ul>
+ * The logtype and dictionary variables are dictionary-encoded, while the encoded variables are stored as longs.
+ * Notably, both {@code encodedVarIds} and {@code encodedVars} are multi-valued, but they are stored using a
+ * flattened single-value mutable forward index, along with a separate forward index to capture the end offsets
+ * for each multi-valued document. This approach is necessary because the maximum number of values per document
+ * is unknown during ingestion, unlike the existing multi-value forward index, which requires this information
+ * upfront. During the conversion from mutable to immutable forward index, the two single-value mutable indices
+ * are merged into a single immutable multi-valued forward index, as the max length is known at conversion time.
+ * <p>
+ * During ingestion, if the cardinality of either the {@code logtypeId} or {@code dictVarID} exceeds a predefined
+ * threshold, the ingestion mode switches to a raw bytes forward index for subsequent documents. Maintaining very
+ * large dictionaries is inefficient in Pinot due to memory and I/O constraints (memory-mapped). Switching to a
+ * raw bytes forward index helps avoid these issues. During reads, if the requested {@code docId} is in the raw
+ * forward index, the raw bytes are returned. Otherwise, the log type, dictionary variables, and encoded variables
+ * are decoded using the CLP decoder to return the original log message's bytes.
+ *
+ * <p><b>Note on Write and Read Operations:</b> Writes are strictly sequential, while reads can be performed
+ * randomly. The supported append operations are:</p>
+ * <ul>
+ * <li>{@code setString(int docId, String value)} - Encodes the log message using CLP and invokes
+ * {@code appendEncodedMessage(@NotNull EncodedMessage clpEncodedMessage)}.</li>
+ * <li>{@code appendEncodedMessage(@NotNull EncodedMessage clpEncodedMessage)}</li>
+ * </ul>
+ *
+ * <p><b>Limitations:</b> The current CLP mutable forward index does not achieve the same compression ratio as the
+ * original standalone CLP implementation, primarily due to design differences between Pinot and CLP. While Pinot
+ * is optimized for fast random access, CLP is designed for single-pass streaming compression and search. As a result,
+ * Pinot sacrifices some compression efficiency for these primary reasons:
+ * <ul>
+ * <li>Pinot implementation uses block compression compared to CLP’s standalone streaming compression, which achieves
+ * much better compression ratio at the cost of random access performance.</li>
+ * <li>Pinot implementation uses uncompressed dictionaries for random lookups of log types and dictionary variables,
+ * whereas CLP employs compressed dictionaries suited only for single-pass streaming queries.</li>
+ * <li>Pinot stores additional offsets and length metadata for log types, dictionary variables, and encoded
+ * variables:</li>
+ * <ul>
+ * <li>Streaming forward indices used by CLP's standalone implementation, do not need to store document start or
+ * end markers because the boundary of one document naturally aligns with the next.</li>
+ * <li>CLP avoids storing the number of {@code dictVars} and {@code encodedVars}, as this information is already
+ * embedded in the log type and available during decoding. Pinot, however, needs to store this metadata,
+ * which can sometimes take up more space than the data itself when compressed.</li>
+ * </ul>
+ * </ul>
+ * Additionally, CLP standalone binaries can perform search without decompressing the data into plain text. The most
+ * common query type on log data is partial matching. Frequently, searches can be completed by scanning only the logtype
+ * dictionaries for partial matches. Searches can also be executed on {@code logtypeId} and {@code dictVarId} directly
+ * by first performing lookup on the dictionaries to get a subset of dictionary ids to filter, whereas in Pinot,
+ * each dictionary id must be first converted back to strings, followed by a brute-force search on the corresponding
+ * string value. For these reasons, direct searching on CLP columns in Pinot is not yet implemented but may be
+ * included in future updates.</p>
+ */
+public class CLPMutableForwardIndexV2 implements MutableForwardIndex {
+ protected static final Logger LOGGER = LoggerFactory.getLogger(CLPMutableForwardIndexV2.class);
+ public final String _columnName;
+
+ protected final EncodedMessage _clpEncodedMessage;
+ protected final MessageEncoder _clpMessageEncoder;
+ protected final MessageDecoder _clpMessageDecoder;
+
+ protected int _nextDocId = 0;
+ protected int _nextDictVarDocId = 0;
+ protected int _nextEncodedVarId = 0;
+ protected int _bytesRawFwdIndexDocIdStartOffset = Integer.MAX_VALUE;
+ protected boolean _isClpEncoded = true;
+ protected int _lengthOfLongestElement;
+ protected int _lengthOfShortestElement;
+
+ protected int _maxNumDictVarIdPerDoc = 0;
+ protected int _maxNumEncodedVarPerDoc = 0;
+
+ protected VarByteSVMutableForwardIndex _rawBytes;
+ protected BytesOffHeapMutableDictionary _logtypeDict;
+ protected FixedByteSVMutableForwardIndex _logtypeId;
+ protected BytesOffHeapMutableDictionary _dictVarDict;
+ protected FixedByteSVMutableForwardIndex _dictVarOffset;
+ protected FixedByteSVMutableForwardIndex _dictVarId;
+ protected FixedByteSVMutableForwardIndex _encodedVarOffset;
+ protected FixedByteSVMutableForwardIndex _encodedVar;
+
+ // Various forward index and dictionary configurations with default values
+ // TODO: Provide more optimized default values in the future
+ protected int _estimatedMaxDocCount = 4096;
+ protected int _rawMessageEstimatedAvgEncodedLength = 256;
+ protected int _estimatedLogtypeAvgEncodedLength = 256;
+ protected int _logtypeIdNumRowsPerChunk = _estimatedMaxDocCount;
+ protected int _logtypeDictEstimatedCardinality = _estimatedMaxDocCount / 16;
+ protected int _dictVarDictEstimatedCardinality = _estimatedMaxDocCount / 8;
+ protected int _logtypeDictMaxOverflowHashSize = 128;
+ protected int _dictVarEstimatedAverageLength = 64;
+ protected int _dictVarOffsetPerChunk = 4 * 1024;
+ protected int _dictVarIdPerChunk = 256 * 1024;
+ protected int _dictVarDictMaxOverflowHashSize = 256;
+ protected int _encodedVarOffsetPerChunk = 4 * 1024;
+ protected int _encodedVarPerChunk = 256 * 1024;
+
+ // Dynamic CLP dictionary encoding configs
+ protected int _minNumDocsBeforeCardinalityMonitoring = _estimatedMaxDocCount / 16;
+ protected boolean _forceEnableDictEncoding = false;
+ protected int _inverseLogtypeCardinalityRatioStopThreshold = 10;
+ protected int _inverseDictVarCardinalityRatioStopThreshold = 10;
+
+ public CLPMutableForwardIndexV2(String columnName, PinotDataBufferMemoryManager memoryManager) {
+ _columnName = columnName;
+
+ // Initialize clp-ffi datastructures
+ _clpEncodedMessage = new EncodedMessage();
+ _clpMessageEncoder = new MessageEncoder(BuiltInVariableHandlingRuleVersions.VariablesSchemaV2,
+ BuiltInVariableHandlingRuleVersions.VariableEncodingMethodsV1);
+ _clpMessageDecoder = new MessageDecoder(BuiltInVariableHandlingRuleVersions.VariablesSchemaV2,
+ BuiltInVariableHandlingRuleVersions.VariableEncodingMethodsV1);
+
+ // Raw forward index stored as bytes
+ _rawBytes = new VarByteSVMutableForwardIndex(FieldSpec.DataType.BYTES, memoryManager, columnName + "_rawBytes.fwd",
+ _estimatedMaxDocCount, _rawMessageEstimatedAvgEncodedLength);
+
+ // LogtypeId + logtype dictionary
+ _logtypeId =
+ new FixedByteSVMutableForwardIndex(false, FieldSpec.DataType.INT, _logtypeIdNumRowsPerChunk, memoryManager,
+ columnName + "_logtypeId.fwd");
+ _logtypeDict = new BytesOffHeapMutableDictionary(_logtypeDictEstimatedCardinality, _logtypeDictMaxOverflowHashSize,
+ memoryManager, columnName + "_logtype.dict", _estimatedLogtypeAvgEncodedLength);
+
+ // DictVars
+ _dictVarOffset =
+ new FixedByteSVMutableForwardIndex(false, FieldSpec.DataType.INT, _dictVarOffsetPerChunk, memoryManager,
+ columnName + "_dictVarOffsets.fwd");
+ _dictVarId = new FixedByteSVMutableForwardIndex(false, FieldSpec.DataType.INT, _dictVarIdPerChunk, memoryManager,
+ columnName + "_dictVarIds.fwd");
+ _dictVarDict = new BytesOffHeapMutableDictionary(_dictVarDictEstimatedCardinality, _dictVarDictMaxOverflowHashSize,
+ memoryManager, columnName + "_dictVar.dict", _dictVarEstimatedAverageLength);
+
+ // EncodedVars
+ _encodedVarOffset =
+ new FixedByteSVMutableForwardIndex(false, FieldSpec.DataType.INT, _encodedVarOffsetPerChunk, memoryManager,
+ columnName + "_encodedVarOffsets.fwd");
+ _encodedVar = new FixedByteSVMutableForwardIndex(false, FieldSpec.DataType.LONG, _encodedVarPerChunk, memoryManager,
+ columnName + "_encodedVar.fwd");
+
+ // Setup offsets used to access flattened dictVarId and encoded var multi-value docs
+ _dictVarOffset.setInt(0, 0);
+ _encodedVarOffset.setInt(0, 0);
+ }
+
+ /**
+ * Sets a string value in the forward index.
+ * <p>
+ * This method appends the given string value to the forward index. The provided `docId` is ignored as this mutable
+ * forward index only supports sequential writes (append operations) rather than random access based on document IDs.
+ * Only reads can be random.
+ *
+ * @param docId The document ID (ignored in this implementation).
+ * @param value The string value to append to the forward index.
+ */
+ @Override
+ public void setString(int docId, String value) {
+ // docId is intentionally ignored because this forward index only supports sequential writes (append only)
+ try {
+ _clpMessageEncoder.encodeMessage(value, _clpEncodedMessage);
+ appendEncodedMessage(_clpEncodedMessage);
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Failed to encode message: " + value, e);
+ }
+ }
+
+ /**
+ * Appends an encoded message to the forward index.
+ * <p>
+ * This method processes the provided {@link EncodedMessage} by first flattening the dictionary variables to ensure
+ * efficient data access through the lower-level clp-ffi API. The method handles potential null values within the
+ * encoded message by replacing them with empty arrays, as Pinot does not accept null values.
+ *
+ * @param clpEncodedMessage The {@link EncodedMessage} to append.
+ * @throws IOException if an I/O error occurs during the appending process.
+ */
+ public void appendEncodedMessage(@NotNull EncodedMessage clpEncodedMessage)
+ throws IOException {
+ if (_isClpEncoded) {
+ // Note: Non-null value is always returned from EncodedMessage.getLogtype()
+ _logtypeId.setInt(_nextDocId, _logtypeDict.index(clpEncodedMessage.getLogtype()));
+
+ // Note: Null value could be returned by Non-null values are always returned from
+ // EncodedMessage.getDictionaryVarsAsFlattenedByteArray(), but
+ // FlattenedByteArray::getDictionaryVarsAsFlattenedByteArray() and FlattenedByteArray::getFlattenedElems()
+ FlattenedByteArray flattenedDictVars =
+ EmptyArrayUtils.getNonNullFlattenedByteArray(clpEncodedMessage.getDictionaryVarsAsFlattenedByteArray());
+ for (byte[] dictVar : flattenedDictVars) {
+ _dictVarId.setInt(_nextDictVarDocId++, _dictVarDict.index(dictVar));
+ }
+ _dictVarOffset.setInt(_nextDocId, _nextDictVarDocId);
+ _maxNumDictVarIdPerDoc = Math.max(_maxNumDictVarIdPerDoc, flattenedDictVars.size());
+
+ // EncodedVars column typically have fairly high cardinality, so we skip dictionary encoding entirely
+ long[] encodedVars = EmptyArrayUtils.getNonNullLongArray(clpEncodedMessage.getEncodedVars());
+ for (long encodedVar : encodedVars) {
+ _encodedVar.setLong(_nextEncodedVarId++, encodedVar);
+ }
+ _encodedVarOffset.setInt(_nextDocId, _nextEncodedVarId);
+ _maxNumEncodedVarPerDoc = Math.max(_maxNumEncodedVarPerDoc, encodedVars.length);
+
+ // Turn off clp encoding when dictionary size is exceeded
+ if (_nextDocId > _minNumDocsBeforeCardinalityMonitoring) {
+ int inverseLogtypeCardinalityRatio = _nextDocId / _logtypeDict.length();
+ if (inverseLogtypeCardinalityRatio < _inverseLogtypeCardinalityRatioStopThreshold) {
+ _isClpEncoded = false;
+ _bytesRawFwdIndexDocIdStartOffset = _nextDocId + 1;
+ } else if (_dictVarDict.length() > 0) {
+ int inverseDictVarCardinalityRatio = _nextDictVarDocId / _dictVarDict.length();
+ if (inverseDictVarCardinalityRatio < _inverseDictVarCardinalityRatioStopThreshold) {
+ _isClpEncoded = false;
+ _bytesRawFwdIndexDocIdStartOffset = _nextDocId + 1;
+ }
+ }
+ }
+ } else {
+ _rawBytes.setBytes(_nextDocId - _bytesRawFwdIndexDocIdStartOffset, clpEncodedMessage.getMessage());
+ }
+ _nextDocId++;
+
+ // Update mutable index statistics for compatibility purposes only
+ _lengthOfLongestElement = Math.max(_lengthOfLongestElement, clpEncodedMessage.getMessage().length);
+ _lengthOfShortestElement = Math.min(_lengthOfShortestElement, clpEncodedMessage.getMessage().length);
+ }
+
+ public int getNumDoc() {
+ return _nextDocId;
+ }
+
+ public int getNumLogtype() {
+ return _isClpEncoded ? _nextDocId : 0;
+ }
+
+ public int getNumDictVar() {
+ return _isClpEncoded ? _nextDictVarDocId : 0;
+ }
+
+ public int getNumEncodedVar() {
+ return _isClpEncoded ? _nextEncodedVarId : 0;
+ }
+
+ /**
+ * Forces the use of CLP dictionary encoding, overriding any automatic encoding decisions.
+ *
+ * <p><b>Note:</b> This method is exclusive to {@code forceRawEncoding}; enabling CLP dictionary
+ * encoding will disable any forced raw encoding. Only one of these methods can be active at a time.</p>
+ */
+ public void forceClpDictionaryEncoding() {
+ _forceEnableDictEncoding = true;
+ }
+
+ /**
+ * Forces the use of raw encoding, overriding any automatic encoding decisions.
+ *
+ * <p><b>Note:</b> This method is exclusive to {@code forceClpDictionaryEncoding}; enabling raw encoding will
+ * disable clp dictionary encoding. Only one of these methods can be active at a time.</p>
+ */
+ public void forceRawEncoding() {
+ _isClpEncoded = false;
+ _bytesRawFwdIndexDocIdStartOffset = 0;
+ }
+
+ public String getColumnName() {
+ return _columnName;
+ }
+
+ @Override
+ public String getString(int docId) {
+ return new String(getRawBytes(docId), StandardCharsets.UTF_8);
+ }
+
+ public byte[] getRawBytes(int docId) {
+ if (docId < 0) {
+ throw new IllegalArgumentException("Invalid docId: " + docId);
+ }
+ if (docId < _bytesRawFwdIndexDocIdStartOffset) {
+ // Decode from clp column
+ byte[] logtype = _logtypeDict.get(_logtypeId.getInt(docId));
+
+ int dictVarIdBeginOffset = (0 == docId) ? 0 : _dictVarOffset.getInt(docId - 1);
+ int dictVarIdEndOffset = _dictVarOffset.getInt(docId);
+ byte[][] dictVars = new byte[dictVarIdEndOffset - dictVarIdBeginOffset][];
+ for (int i = 0; i < dictVars.length; i++) {
+ dictVars[i] = _dictVarDict.get(_dictVarId.getInt(dictVarIdBeginOffset + i));
+ }
+ FlattenedByteArray flattenedDictVars = FlattenedByteArrayFactory.fromByteArrays(dictVars);
+
+ int encodedVarIdBeginOffset = (0 == docId) ? 0 : _encodedVarOffset.getInt(docId - 1);
+ int encodedVarIdEndOffset = _encodedVarOffset.getInt(docId);
+ long[] encodedVars = new long[encodedVarIdEndOffset - encodedVarIdBeginOffset];
+ for (int i = 0; i < encodedVars.length; i++) {
+ encodedVars[i] = _encodedVar.getLong(encodedVarIdBeginOffset + i);
+ }
+
+ try {
+ return _clpMessageDecoder.decodeMessageAsBytes(logtype, flattenedDictVars, encodedVars);
+ } catch (IOException e) {
+ throw new IllegalArgumentException(
+ "Failed to encode message: " + new String(logtype, StandardCharsets.ISO_8859_1), e);
+ }
+ } else {
+ return _rawBytes.getBytes(docId - _bytesRawFwdIndexDocIdStartOffset);
+ }
+ }
+
+ public FixedByteSVMutableForwardIndex getLogtypeId() {
+ return _logtypeId;
+ }
+
+ public BytesOffHeapMutableDictionary getLogtypeDict() {
+ return _logtypeDict;
+ }
+
+ public FixedByteSVMutableForwardIndex getDictVarOffset() {
+ return _dictVarOffset;
+ }
+
+ public FixedByteSVMutableForwardIndex getDictVarId() {
+ return _dictVarId;
+ }
+
+ public BytesOffHeapMutableDictionary getDictVarDict() {
+ return _dictVarDict;
+ }
+
+ public FixedByteSVMutableForwardIndex getEncodedVarOffset() {
+ return _encodedVarOffset;
+ }
+
+ public FixedByteSVMutableForwardIndex getEncodedVar() {
+ return _encodedVar;
+ }
+
+ /**
+ * Returns whether the mutable forward index is currently using dictionary encoding.
Review Comment:
> whether the mutable forward index is currently using dictionary encoding
whether the mutable forward index is currently using CLP forward indexes with dictionary encoding or rawFwdIndex.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/CLPMutableForwardIndexV2.java:
##########
@@ -0,0 +1,468 @@
+/**
+ * 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.segment.local.realtime.impl.forward;
+
+import com.yscope.clp.compressorfrontend.BuiltInVariableHandlingRuleVersions;
+import com.yscope.clp.compressorfrontend.EmptyArrayUtils;
+import com.yscope.clp.compressorfrontend.EncodedMessage;
+import com.yscope.clp.compressorfrontend.FlattenedByteArray;
+import com.yscope.clp.compressorfrontend.FlattenedByteArrayFactory;
+import com.yscope.clp.compressorfrontend.MessageDecoder;
+import com.yscope.clp.compressorfrontend.MessageEncoder;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import javax.validation.constraints.NotNull;
+import org.apache.pinot.segment.local.realtime.impl.dictionary.BytesOffHeapMutableDictionary;
+import org.apache.pinot.segment.spi.index.mutable.MutableForwardIndex;
+import org.apache.pinot.segment.spi.memory.PinotDataBufferMemoryManager;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This mutable forward index implements a composite index for string-typed columns with dynamic encoding options:
+ * <ol>
+ * <li>Pure CLP dictionary encoding when the dictionary cardinality is below a configurable threshold.</li>
+ * <li>CLP dictionary encoding combined with a raw string forward index when the dictionary cardinality exceeds
+ * the threshold.</li>
+ * </ol>
+ * <p>
+ * Initially, CLP encoding transforms a high-cardinality log message string into three data columns:
+ * <ul>
+ * <li>Logtype (very low cardinality) - essentially an inferred format string of the log</li>
+ * <li>Dictionary variables (medium cardinality) - variables with both alphabets and numbers</li>
+ * <li>Encoded variables (high cardinality) - pure fixed point and floating point numbers</li>
+ * </ul>
+ * The logtype and dictionary variables are dictionary-encoded, while the encoded variables are stored as longs.
+ * Notably, both {@code encodedVarIds} and {@code encodedVars} are multi-valued, but they are stored using a
+ * flattened single-value mutable forward index, along with a separate forward index to capture the end offsets
+ * for each multi-valued document. This approach is necessary because the maximum number of values per document
+ * is unknown during ingestion, unlike the existing multi-value forward index, which requires this information
+ * upfront. During the conversion from mutable to immutable forward index, the two single-value mutable indices
+ * are merged into a single immutable multi-valued forward index, as the max length is known at conversion time.
+ * <p>
+ * During ingestion, if the cardinality of either the {@code logtypeId} or {@code dictVarID} exceeds a predefined
+ * threshold, the ingestion mode switches to a raw bytes forward index for subsequent documents. Maintaining very
+ * large dictionaries is inefficient in Pinot due to memory and I/O constraints (memory-mapped). Switching to a
+ * raw bytes forward index helps avoid these issues. During reads, if the requested {@code docId} is in the raw
+ * forward index, the raw bytes are returned. Otherwise, the log type, dictionary variables, and encoded variables
+ * are decoded using the CLP decoder to return the original log message's bytes.
+ *
+ * <p><b>Note on Write and Read Operations:</b> Writes are strictly sequential, while reads can be performed
+ * randomly. The supported append operations are:</p>
+ * <ul>
+ * <li>{@code setString(int docId, String value)} - Encodes the log message using CLP and invokes
+ * {@code appendEncodedMessage(@NotNull EncodedMessage clpEncodedMessage)}.</li>
+ * <li>{@code appendEncodedMessage(@NotNull EncodedMessage clpEncodedMessage)}</li>
+ * </ul>
+ *
+ * <p><b>Limitations:</b> The current CLP mutable forward index does not achieve the same compression ratio as the
+ * original standalone CLP implementation, primarily due to design differences between Pinot and CLP. While Pinot
+ * is optimized for fast random access, CLP is designed for single-pass streaming compression and search. As a result,
+ * Pinot sacrifices some compression efficiency for these primary reasons:
+ * <ul>
+ * <li>Pinot implementation uses block compression compared to CLP’s standalone streaming compression, which achieves
+ * much better compression ratio at the cost of random access performance.</li>
+ * <li>Pinot implementation uses uncompressed dictionaries for random lookups of log types and dictionary variables,
+ * whereas CLP employs compressed dictionaries suited only for single-pass streaming queries.</li>
+ * <li>Pinot stores additional offsets and length metadata for log types, dictionary variables, and encoded
+ * variables:</li>
+ * <ul>
+ * <li>Streaming forward indices used by CLP's standalone implementation, do not need to store document start or
+ * end markers because the boundary of one document naturally aligns with the next.</li>
+ * <li>CLP avoids storing the number of {@code dictVars} and {@code encodedVars}, as this information is already
+ * embedded in the log type and available during decoding. Pinot, however, needs to store this metadata,
+ * which can sometimes take up more space than the data itself when compressed.</li>
+ * </ul>
+ * </ul>
+ * Additionally, CLP standalone binaries can perform search without decompressing the data into plain text. The most
+ * common query type on log data is partial matching. Frequently, searches can be completed by scanning only the logtype
+ * dictionaries for partial matches. Searches can also be executed on {@code logtypeId} and {@code dictVarId} directly
+ * by first performing lookup on the dictionaries to get a subset of dictionary ids to filter, whereas in Pinot,
+ * each dictionary id must be first converted back to strings, followed by a brute-force search on the corresponding
+ * string value. For these reasons, direct searching on CLP columns in Pinot is not yet implemented but may be
+ * included in future updates.</p>
+ */
+public class CLPMutableForwardIndexV2 implements MutableForwardIndex {
+ protected static final Logger LOGGER = LoggerFactory.getLogger(CLPMutableForwardIndexV2.class);
+ public final String _columnName;
+
+ protected final EncodedMessage _clpEncodedMessage;
+ protected final MessageEncoder _clpMessageEncoder;
+ protected final MessageDecoder _clpMessageDecoder;
+
+ protected int _nextDocId = 0;
+ protected int _nextDictVarDocId = 0;
+ protected int _nextEncodedVarId = 0;
+ protected int _bytesRawFwdIndexDocIdStartOffset = Integer.MAX_VALUE;
+ protected boolean _isClpEncoded = true;
+ protected int _lengthOfLongestElement;
+ protected int _lengthOfShortestElement;
+
+ protected int _maxNumDictVarIdPerDoc = 0;
+ protected int _maxNumEncodedVarPerDoc = 0;
+
+ protected VarByteSVMutableForwardIndex _rawBytes;
+ protected BytesOffHeapMutableDictionary _logtypeDict;
+ protected FixedByteSVMutableForwardIndex _logtypeId;
+ protected BytesOffHeapMutableDictionary _dictVarDict;
+ protected FixedByteSVMutableForwardIndex _dictVarOffset;
+ protected FixedByteSVMutableForwardIndex _dictVarId;
+ protected FixedByteSVMutableForwardIndex _encodedVarOffset;
+ protected FixedByteSVMutableForwardIndex _encodedVar;
+
+ // Various forward index and dictionary configurations with default values
+ // TODO: Provide more optimized default values in the future
+ protected int _estimatedMaxDocCount = 4096;
+ protected int _rawMessageEstimatedAvgEncodedLength = 256;
+ protected int _estimatedLogtypeAvgEncodedLength = 256;
+ protected int _logtypeIdNumRowsPerChunk = _estimatedMaxDocCount;
+ protected int _logtypeDictEstimatedCardinality = _estimatedMaxDocCount / 16;
+ protected int _dictVarDictEstimatedCardinality = _estimatedMaxDocCount / 8;
+ protected int _logtypeDictMaxOverflowHashSize = 128;
+ protected int _dictVarEstimatedAverageLength = 64;
+ protected int _dictVarOffsetPerChunk = 4 * 1024;
+ protected int _dictVarIdPerChunk = 256 * 1024;
+ protected int _dictVarDictMaxOverflowHashSize = 256;
+ protected int _encodedVarOffsetPerChunk = 4 * 1024;
+ protected int _encodedVarPerChunk = 256 * 1024;
+
+ // Dynamic CLP dictionary encoding configs
+ protected int _minNumDocsBeforeCardinalityMonitoring = _estimatedMaxDocCount / 16;
+ protected boolean _forceEnableDictEncoding = false;
+ protected int _inverseLogtypeCardinalityRatioStopThreshold = 10;
+ protected int _inverseDictVarCardinalityRatioStopThreshold = 10;
+
+ public CLPMutableForwardIndexV2(String columnName, PinotDataBufferMemoryManager memoryManager) {
+ _columnName = columnName;
+
+ // Initialize clp-ffi datastructures
+ _clpEncodedMessage = new EncodedMessage();
+ _clpMessageEncoder = new MessageEncoder(BuiltInVariableHandlingRuleVersions.VariablesSchemaV2,
+ BuiltInVariableHandlingRuleVersions.VariableEncodingMethodsV1);
+ _clpMessageDecoder = new MessageDecoder(BuiltInVariableHandlingRuleVersions.VariablesSchemaV2,
+ BuiltInVariableHandlingRuleVersions.VariableEncodingMethodsV1);
+
+ // Raw forward index stored as bytes
+ _rawBytes = new VarByteSVMutableForwardIndex(FieldSpec.DataType.BYTES, memoryManager, columnName + "_rawBytes.fwd",
+ _estimatedMaxDocCount, _rawMessageEstimatedAvgEncodedLength);
+
+ // LogtypeId + logtype dictionary
+ _logtypeId =
+ new FixedByteSVMutableForwardIndex(false, FieldSpec.DataType.INT, _logtypeIdNumRowsPerChunk, memoryManager,
+ columnName + "_logtypeId.fwd");
+ _logtypeDict = new BytesOffHeapMutableDictionary(_logtypeDictEstimatedCardinality, _logtypeDictMaxOverflowHashSize,
+ memoryManager, columnName + "_logtype.dict", _estimatedLogtypeAvgEncodedLength);
+
+ // DictVars
+ _dictVarOffset =
+ new FixedByteSVMutableForwardIndex(false, FieldSpec.DataType.INT, _dictVarOffsetPerChunk, memoryManager,
+ columnName + "_dictVarOffsets.fwd");
+ _dictVarId = new FixedByteSVMutableForwardIndex(false, FieldSpec.DataType.INT, _dictVarIdPerChunk, memoryManager,
+ columnName + "_dictVarIds.fwd");
+ _dictVarDict = new BytesOffHeapMutableDictionary(_dictVarDictEstimatedCardinality, _dictVarDictMaxOverflowHashSize,
+ memoryManager, columnName + "_dictVar.dict", _dictVarEstimatedAverageLength);
+
+ // EncodedVars
+ _encodedVarOffset =
+ new FixedByteSVMutableForwardIndex(false, FieldSpec.DataType.INT, _encodedVarOffsetPerChunk, memoryManager,
+ columnName + "_encodedVarOffsets.fwd");
+ _encodedVar = new FixedByteSVMutableForwardIndex(false, FieldSpec.DataType.LONG, _encodedVarPerChunk, memoryManager,
+ columnName + "_encodedVar.fwd");
+
+ // Setup offsets used to access flattened dictVarId and encoded var multi-value docs
+ _dictVarOffset.setInt(0, 0);
+ _encodedVarOffset.setInt(0, 0);
+ }
+
+ /**
+ * Sets a string value in the forward index.
+ * <p>
+ * This method appends the given string value to the forward index. The provided `docId` is ignored as this mutable
+ * forward index only supports sequential writes (append operations) rather than random access based on document IDs.
+ * Only reads can be random.
+ *
+ * @param docId The document ID (ignored in this implementation).
+ * @param value The string value to append to the forward index.
+ */
+ @Override
+ public void setString(int docId, String value) {
+ // docId is intentionally ignored because this forward index only supports sequential writes (append only)
+ try {
+ _clpMessageEncoder.encodeMessage(value, _clpEncodedMessage);
+ appendEncodedMessage(_clpEncodedMessage);
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Failed to encode message: " + value, e);
+ }
+ }
+
+ /**
+ * Appends an encoded message to the forward index.
+ * <p>
+ * This method processes the provided {@link EncodedMessage} by first flattening the dictionary variables to ensure
+ * efficient data access through the lower-level clp-ffi API. The method handles potential null values within the
+ * encoded message by replacing them with empty arrays, as Pinot does not accept null values.
+ *
+ * @param clpEncodedMessage The {@link EncodedMessage} to append.
+ * @throws IOException if an I/O error occurs during the appending process.
+ */
+ public void appendEncodedMessage(@NotNull EncodedMessage clpEncodedMessage)
+ throws IOException {
+ if (_isClpEncoded) {
+ // Note: Non-null value is always returned from EncodedMessage.getLogtype()
+ _logtypeId.setInt(_nextDocId, _logtypeDict.index(clpEncodedMessage.getLogtype()));
+
+ // Note: Null value could be returned by Non-null values are always returned from
+ // EncodedMessage.getDictionaryVarsAsFlattenedByteArray(), but
+ // FlattenedByteArray::getDictionaryVarsAsFlattenedByteArray() and FlattenedByteArray::getFlattenedElems()
+ FlattenedByteArray flattenedDictVars =
+ EmptyArrayUtils.getNonNullFlattenedByteArray(clpEncodedMessage.getDictionaryVarsAsFlattenedByteArray());
+ for (byte[] dictVar : flattenedDictVars) {
+ _dictVarId.setInt(_nextDictVarDocId++, _dictVarDict.index(dictVar));
+ }
+ _dictVarOffset.setInt(_nextDocId, _nextDictVarDocId);
+ _maxNumDictVarIdPerDoc = Math.max(_maxNumDictVarIdPerDoc, flattenedDictVars.size());
+
+ // EncodedVars column typically have fairly high cardinality, so we skip dictionary encoding entirely
+ long[] encodedVars = EmptyArrayUtils.getNonNullLongArray(clpEncodedMessage.getEncodedVars());
+ for (long encodedVar : encodedVars) {
+ _encodedVar.setLong(_nextEncodedVarId++, encodedVar);
+ }
+ _encodedVarOffset.setInt(_nextDocId, _nextEncodedVarId);
+ _maxNumEncodedVarPerDoc = Math.max(_maxNumEncodedVarPerDoc, encodedVars.length);
+
+ // Turn off clp encoding when dictionary size is exceeded
+ if (_nextDocId > _minNumDocsBeforeCardinalityMonitoring) {
+ int inverseLogtypeCardinalityRatio = _nextDocId / _logtypeDict.length();
+ if (inverseLogtypeCardinalityRatio < _inverseLogtypeCardinalityRatioStopThreshold) {
+ _isClpEncoded = false;
+ _bytesRawFwdIndexDocIdStartOffset = _nextDocId + 1;
+ } else if (_dictVarDict.length() > 0) {
+ int inverseDictVarCardinalityRatio = _nextDictVarDocId / _dictVarDict.length();
+ if (inverseDictVarCardinalityRatio < _inverseDictVarCardinalityRatioStopThreshold) {
+ _isClpEncoded = false;
+ _bytesRawFwdIndexDocIdStartOffset = _nextDocId + 1;
+ }
+ }
+ }
+ } else {
+ _rawBytes.setBytes(_nextDocId - _bytesRawFwdIndexDocIdStartOffset, clpEncodedMessage.getMessage());
+ }
+ _nextDocId++;
+
+ // Update mutable index statistics for compatibility purposes only
+ _lengthOfLongestElement = Math.max(_lengthOfLongestElement, clpEncodedMessage.getMessage().length);
+ _lengthOfShortestElement = Math.min(_lengthOfShortestElement, clpEncodedMessage.getMessage().length);
+ }
+
+ public int getNumDoc() {
+ return _nextDocId;
+ }
+
+ public int getNumLogtype() {
+ return _isClpEncoded ? _nextDocId : 0;
+ }
+
+ public int getNumDictVar() {
+ return _isClpEncoded ? _nextDictVarDocId : 0;
+ }
+
+ public int getNumEncodedVar() {
+ return _isClpEncoded ? _nextEncodedVarId : 0;
+ }
+
+ /**
+ * Forces the use of CLP dictionary encoding, overriding any automatic encoding decisions.
+ *
+ * <p><b>Note:</b> This method is exclusive to {@code forceRawEncoding}; enabling CLP dictionary
+ * encoding will disable any forced raw encoding. Only one of these methods can be active at a time.</p>
+ */
+ public void forceClpDictionaryEncoding() {
+ _forceEnableDictEncoding = true;
+ }
+
+ /**
+ * Forces the use of raw encoding, overriding any automatic encoding decisions.
+ *
+ * <p><b>Note:</b> This method is exclusive to {@code forceClpDictionaryEncoding}; enabling raw encoding will
+ * disable clp dictionary encoding. Only one of these methods can be active at a time.</p>
+ */
+ public void forceRawEncoding() {
+ _isClpEncoded = false;
+ _bytesRawFwdIndexDocIdStartOffset = 0;
+ }
+
+ public String getColumnName() {
+ return _columnName;
+ }
+
+ @Override
+ public String getString(int docId) {
+ return new String(getRawBytes(docId), StandardCharsets.UTF_8);
+ }
+
+ public byte[] getRawBytes(int docId) {
+ if (docId < 0) {
+ throw new IllegalArgumentException("Invalid docId: " + docId);
+ }
+ if (docId < _bytesRawFwdIndexDocIdStartOffset) {
+ // Decode from clp column
+ byte[] logtype = _logtypeDict.get(_logtypeId.getInt(docId));
+
+ int dictVarIdBeginOffset = (0 == docId) ? 0 : _dictVarOffset.getInt(docId - 1);
+ int dictVarIdEndOffset = _dictVarOffset.getInt(docId);
+ byte[][] dictVars = new byte[dictVarIdEndOffset - dictVarIdBeginOffset][];
+ for (int i = 0; i < dictVars.length; i++) {
+ dictVars[i] = _dictVarDict.get(_dictVarId.getInt(dictVarIdBeginOffset + i));
+ }
+ FlattenedByteArray flattenedDictVars = FlattenedByteArrayFactory.fromByteArrays(dictVars);
+
+ int encodedVarIdBeginOffset = (0 == docId) ? 0 : _encodedVarOffset.getInt(docId - 1);
+ int encodedVarIdEndOffset = _encodedVarOffset.getInt(docId);
+ long[] encodedVars = new long[encodedVarIdEndOffset - encodedVarIdBeginOffset];
+ for (int i = 0; i < encodedVars.length; i++) {
+ encodedVars[i] = _encodedVar.getLong(encodedVarIdBeginOffset + i);
+ }
+
+ try {
+ return _clpMessageDecoder.decodeMessageAsBytes(logtype, flattenedDictVars, encodedVars);
+ } catch (IOException e) {
+ throw new IllegalArgumentException(
+ "Failed to encode message: " + new String(logtype, StandardCharsets.ISO_8859_1), e);
+ }
+ } else {
+ return _rawBytes.getBytes(docId - _bytesRawFwdIndexDocIdStartOffset);
+ }
+ }
+
+ public FixedByteSVMutableForwardIndex getLogtypeId() {
+ return _logtypeId;
+ }
+
+ public BytesOffHeapMutableDictionary getLogtypeDict() {
+ return _logtypeDict;
+ }
+
+ public FixedByteSVMutableForwardIndex getDictVarOffset() {
+ return _dictVarOffset;
+ }
+
+ public FixedByteSVMutableForwardIndex getDictVarId() {
+ return _dictVarId;
+ }
+
+ public BytesOffHeapMutableDictionary getDictVarDict() {
+ return _dictVarDict;
+ }
+
+ public FixedByteSVMutableForwardIndex getEncodedVarOffset() {
+ return _encodedVarOffset;
+ }
+
+ public FixedByteSVMutableForwardIndex getEncodedVar() {
+ return _encodedVar;
+ }
+
+ /**
+ * Returns whether the mutable forward index is currently using dictionary encoding.
Review Comment:
> whether the mutable forward index is currently using dictionary encoding
change to
whether the mutable forward index is currently using CLP forward indexes with dictionary encoding or rawFwdIndex.
--
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
Re: [PR] Add the initial implementation of CLPMutableForwardIndexV2 [pinot]
Posted by "jackluo923 (via GitHub)" <gi...@apache.org>.
jackluo923 commented on code in PR #14241:
URL: https://github.com/apache/pinot/pull/14241#discussion_r1802042000
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/CLPMutableForwardIndexV2.java:
##########
@@ -0,0 +1,468 @@
+/**
+ * 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.segment.local.realtime.impl.forward;
+
+import com.yscope.clp.compressorfrontend.BuiltInVariableHandlingRuleVersions;
+import com.yscope.clp.compressorfrontend.EmptyArrayUtils;
+import com.yscope.clp.compressorfrontend.EncodedMessage;
+import com.yscope.clp.compressorfrontend.FlattenedByteArray;
+import com.yscope.clp.compressorfrontend.FlattenedByteArrayFactory;
+import com.yscope.clp.compressorfrontend.MessageDecoder;
+import com.yscope.clp.compressorfrontend.MessageEncoder;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import javax.validation.constraints.NotNull;
+import org.apache.pinot.segment.local.realtime.impl.dictionary.BytesOffHeapMutableDictionary;
+import org.apache.pinot.segment.spi.index.mutable.MutableForwardIndex;
+import org.apache.pinot.segment.spi.memory.PinotDataBufferMemoryManager;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This mutable forward index implements a composite index for string-typed columns with dynamic encoding options:
+ * <ol>
+ * <li>Pure CLP dictionary encoding when the dictionary cardinality is below a configurable threshold.</li>
+ * <li>CLP dictionary encoding combined with a raw string forward index when the dictionary cardinality exceeds
+ * the threshold.</li>
+ * </ol>
+ * <p>
+ * Initially, CLP encoding transforms a high-cardinality log message string into three data columns:
+ * <ul>
+ * <li>Logtype (very low cardinality) - essentially an inferred format string of the log</li>
+ * <li>Dictionary variables (medium cardinality) - variables with both alphabets and numbers</li>
+ * <li>Encoded variables (high cardinality) - pure fixed point and floating point numbers</li>
+ * </ul>
+ * The logtype and dictionary variables are dictionary-encoded, while the encoded variables are stored as longs.
+ * Notably, both {@code encodedVarIds} and {@code encodedVars} are multi-valued, but they are stored using a
+ * flattened single-value mutable forward index, along with a separate forward index to capture the end offsets
+ * for each multi-valued document. This approach is necessary because the maximum number of values per document
+ * is unknown during ingestion, unlike the existing multi-value forward index, which requires this information
+ * upfront. During the conversion from mutable to immutable forward index, the two single-value mutable indices
+ * are merged into a single immutable multi-valued forward index, as the max length is known at conversion time.
+ * <p>
+ * During ingestion, if the cardinality of either the {@code logtypeId} or {@code dictVarID} exceeds a predefined
+ * threshold, the ingestion mode switches to a raw bytes forward index for subsequent documents. Maintaining very
+ * large dictionaries is inefficient in Pinot due to memory and I/O constraints (memory-mapped). Switching to a
+ * raw bytes forward index helps avoid these issues. During reads, if the requested {@code docId} is in the raw
+ * forward index, the raw bytes are returned. Otherwise, the log type, dictionary variables, and encoded variables
+ * are decoded using the CLP decoder to return the original log message's bytes.
+ *
+ * <p><b>Note on Write and Read Operations:</b> Writes are strictly sequential, while reads can be performed
+ * randomly. The supported append operations are:</p>
+ * <ul>
+ * <li>{@code setString(int docId, String value)} - Encodes the log message using CLP and invokes
+ * {@code appendEncodedMessage(@NotNull EncodedMessage clpEncodedMessage)}.</li>
+ * <li>{@code appendEncodedMessage(@NotNull EncodedMessage clpEncodedMessage)}</li>
+ * </ul>
+ *
+ * <p><b>Limitations:</b> The current CLP mutable forward index does not achieve the same compression ratio as the
+ * original standalone CLP implementation, primarily due to design differences between Pinot and CLP. While Pinot
+ * is optimized for fast random access, CLP is designed for single-pass streaming compression and search. As a result,
+ * Pinot sacrifices some compression efficiency for these primary reasons:
+ * <ul>
+ * <li>Pinot implementation uses block compression compared to CLP’s standalone streaming compression, which achieves
+ * much better compression ratio at the cost of random access performance.</li>
+ * <li>Pinot implementation uses uncompressed dictionaries for random lookups of log types and dictionary variables,
+ * whereas CLP employs compressed dictionaries suited only for single-pass streaming queries.</li>
+ * <li>Pinot stores additional offsets and length metadata for log types, dictionary variables, and encoded
+ * variables:</li>
+ * <ul>
+ * <li>Streaming forward indices used by CLP's standalone implementation, do not need to store document start or
+ * end markers because the boundary of one document naturally aligns with the next.</li>
+ * <li>CLP avoids storing the number of {@code dictVars} and {@code encodedVars}, as this information is already
+ * embedded in the log type and available during decoding. Pinot, however, needs to store this metadata,
+ * which can sometimes take up more space than the data itself when compressed.</li>
+ * </ul>
+ * </ul>
+ * Additionally, CLP standalone binaries can perform search without decompressing the data into plain text. The most
+ * common query type on log data is partial matching. Frequently, searches can be completed by scanning only the logtype
+ * dictionaries for partial matches. Searches can also be executed on {@code logtypeId} and {@code dictVarId} directly
+ * by first performing lookup on the dictionaries to get a subset of dictionary ids to filter, whereas in Pinot,
+ * each dictionary id must be first converted back to strings, followed by a brute-force search on the corresponding
+ * string value. For these reasons, direct searching on CLP columns in Pinot is not yet implemented but may be
+ * included in future updates.</p>
+ */
+public class CLPMutableForwardIndexV2 implements MutableForwardIndex {
+ protected static final Logger LOGGER = LoggerFactory.getLogger(CLPMutableForwardIndexV2.class);
+ public final String _columnName;
+
+ protected final EncodedMessage _clpEncodedMessage;
+ protected final MessageEncoder _clpMessageEncoder;
+ protected final MessageDecoder _clpMessageDecoder;
+
+ protected int _nextDocId = 0;
+ protected int _nextDictVarDocId = 0;
+ protected int _nextEncodedVarId = 0;
+ protected int _bytesRawFwdIndexDocIdStartOffset = Integer.MAX_VALUE;
Review Comment:
The value is updated at the end of `appendEncodedMessage` method call. I will add a comment to make this clear.
--
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
Re: [PR] Add and use CLPMutableForwardIndexV2 by default to improve ingestion performance and efficiency [pinot]
Posted by "jackluo923 (via GitHub)" <gi...@apache.org>.
jackluo923 commented on PR #14241:
URL: https://github.com/apache/pinot/pull/14241#issuecomment-2432303780
> For mutable index, v1 and v2 cannot co-exist. If we are not planning to make it configurable, let's just remove v1 and only keep one mutable index format
Sure, I will make the change in another PR.
--
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
Re: [PR] Add and use CLPMutableForwardIndexV2 by default to improve ingestion performance and efficiency [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu merged PR #14241:
URL: https://github.com/apache/pinot/pull/14241
--
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
Re: [PR] Add the initial implementation of CLPMutableForwardIndexV2 [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14241:
URL: https://github.com/apache/pinot/pull/14241#discussion_r1801971635
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/CLPMutableForwardIndexV2.java:
##########
@@ -0,0 +1,468 @@
+/**
+ * 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.segment.local.realtime.impl.forward;
+
+import com.yscope.clp.compressorfrontend.BuiltInVariableHandlingRuleVersions;
+import com.yscope.clp.compressorfrontend.EmptyArrayUtils;
+import com.yscope.clp.compressorfrontend.EncodedMessage;
+import com.yscope.clp.compressorfrontend.FlattenedByteArray;
+import com.yscope.clp.compressorfrontend.FlattenedByteArrayFactory;
+import com.yscope.clp.compressorfrontend.MessageDecoder;
+import com.yscope.clp.compressorfrontend.MessageEncoder;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import javax.validation.constraints.NotNull;
+import org.apache.pinot.segment.local.realtime.impl.dictionary.BytesOffHeapMutableDictionary;
+import org.apache.pinot.segment.spi.index.mutable.MutableForwardIndex;
+import org.apache.pinot.segment.spi.memory.PinotDataBufferMemoryManager;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This mutable forward index implements a composite index for string-typed columns with dynamic encoding options:
+ * <ol>
+ * <li>Pure CLP dictionary encoding when the dictionary cardinality is below a configurable threshold.</li>
+ * <li>CLP dictionary encoding combined with a raw string forward index when the dictionary cardinality exceeds
+ * the threshold.</li>
+ * </ol>
+ * <p>
+ * Initially, CLP encoding transforms a high-cardinality log message string into three data columns:
+ * <ul>
+ * <li>Logtype (very low cardinality) - essentially an inferred format string of the log</li>
+ * <li>Dictionary variables (medium cardinality) - variables with both alphabets and numbers</li>
+ * <li>Encoded variables (high cardinality) - pure fixed point and floating point numbers</li>
+ * </ul>
+ * The logtype and dictionary variables are dictionary-encoded, while the encoded variables are stored as longs.
+ * Notably, both {@code encodedVarIds} and {@code encodedVars} are multi-valued, but they are stored using a
+ * flattened single-value mutable forward index, along with a separate forward index to capture the end offsets
+ * for each multi-valued document. This approach is necessary because the maximum number of values per document
+ * is unknown during ingestion, unlike the existing multi-value forward index, which requires this information
+ * upfront. During the conversion from mutable to immutable forward index, the two single-value mutable indices
+ * are merged into a single immutable multi-valued forward index, as the max length is known at conversion time.
+ * <p>
+ * During ingestion, if the cardinality of either the {@code logtypeId} or {@code dictVarID} exceeds a predefined
+ * threshold, the ingestion mode switches to a raw bytes forward index for subsequent documents. Maintaining very
+ * large dictionaries is inefficient in Pinot due to memory and I/O constraints (memory-mapped). Switching to a
+ * raw bytes forward index helps avoid these issues. During reads, if the requested {@code docId} is in the raw
+ * forward index, the raw bytes are returned. Otherwise, the log type, dictionary variables, and encoded variables
+ * are decoded using the CLP decoder to return the original log message's bytes.
+ *
+ * <p><b>Note on Write and Read Operations:</b> Writes are strictly sequential, while reads can be performed
+ * randomly. The supported append operations are:</p>
+ * <ul>
+ * <li>{@code setString(int docId, String value)} - Encodes the log message using CLP and invokes
+ * {@code appendEncodedMessage(@NotNull EncodedMessage clpEncodedMessage)}.</li>
+ * <li>{@code appendEncodedMessage(@NotNull EncodedMessage clpEncodedMessage)}</li>
+ * </ul>
+ *
+ * <p><b>Limitations:</b> The current CLP mutable forward index does not achieve the same compression ratio as the
+ * original standalone CLP implementation, primarily due to design differences between Pinot and CLP. While Pinot
+ * is optimized for fast random access, CLP is designed for single-pass streaming compression and search. As a result,
+ * Pinot sacrifices some compression efficiency for these primary reasons:
+ * <ul>
+ * <li>Pinot implementation uses block compression compared to CLP’s standalone streaming compression, which achieves
+ * much better compression ratio at the cost of random access performance.</li>
+ * <li>Pinot implementation uses uncompressed dictionaries for random lookups of log types and dictionary variables,
+ * whereas CLP employs compressed dictionaries suited only for single-pass streaming queries.</li>
+ * <li>Pinot stores additional offsets and length metadata for log types, dictionary variables, and encoded
+ * variables:</li>
+ * <ul>
+ * <li>Streaming forward indices used by CLP's standalone implementation, do not need to store document start or
+ * end markers because the boundary of one document naturally aligns with the next.</li>
+ * <li>CLP avoids storing the number of {@code dictVars} and {@code encodedVars}, as this information is already
+ * embedded in the log type and available during decoding. Pinot, however, needs to store this metadata,
+ * which can sometimes take up more space than the data itself when compressed.</li>
+ * </ul>
+ * </ul>
+ * Additionally, CLP standalone binaries can perform search without decompressing the data into plain text. The most
+ * common query type on log data is partial matching. Frequently, searches can be completed by scanning only the logtype
+ * dictionaries for partial matches. Searches can also be executed on {@code logtypeId} and {@code dictVarId} directly
+ * by first performing lookup on the dictionaries to get a subset of dictionary ids to filter, whereas in Pinot,
+ * each dictionary id must be first converted back to strings, followed by a brute-force search on the corresponding
+ * string value. For these reasons, direct searching on CLP columns in Pinot is not yet implemented but may be
+ * included in future updates.</p>
+ */
+public class CLPMutableForwardIndexV2 implements MutableForwardIndex {
+ protected static final Logger LOGGER = LoggerFactory.getLogger(CLPMutableForwardIndexV2.class);
+ public final String _columnName;
+
+ protected final EncodedMessage _clpEncodedMessage;
+ protected final MessageEncoder _clpMessageEncoder;
+ protected final MessageDecoder _clpMessageDecoder;
+
+ protected int _nextDocId = 0;
+ protected int _nextDictVarDocId = 0;
+ protected int _nextEncodedVarId = 0;
+ protected int _bytesRawFwdIndexDocIdStartOffset = Integer.MAX_VALUE;
Review Comment:
is this only for CLPV2? can we add a comment on how this value is dynamic updated?
--
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