Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,9 @@
/// 3. JSON index (JsonMatchFilterOperator)
/// 4. Expression scan fallback (ExpressionFilterOperator)
public class MapFilterOperator extends BaseFilterOperator {
/// A key's JSON path, in the form an index's configured paths are written in ("$.user_tier").
private static final String JSON_PATH_PREFIX = "$.";

private static final String EXPLAIN_NAME = "FILTER_MAP";

private enum DelegateType {
Expand Down Expand Up @@ -161,7 +164,7 @@ private BaseFilterOperator buildPerKeyFilterOperator(DataSource keyDs, QueryCont
private BaseFilterOperator trySparseJsonIndex(OpenStructDataSource osDs, DataSource sparseKeyDs,
QueryContext queryContext, int numDocs) {
JsonIndexReader jsonIndex = osDs.getSparseJsonIndex();
if (jsonIndex == null) {
if (jsonIndex == null || !servesKey(jsonIndex)) {
return null;
}
if (sparseKeyDs.getDataSourceMetadata().getDataType().getStoredType() != FieldSpec.DataType.STRING) {
Expand Down Expand Up @@ -250,13 +253,23 @@ private JsonMatchFilterOperator tryJsonIndex(@Nullable DataSource dataSource, in
jsonIndex = (JsonIndexReader) dataSource.getIndex(compositeIndex.get());
}
}
if (jsonIndex == null) {
if (jsonIndex == null || !servesKey(jsonIndex)) {
return null;
}
FilterContext filterContext = createFilterContext();
return new JsonMatchFilterOperator(jsonIndex, filterContext, numDocs);
}

/// Whether this index actually holds postings for the key being filtered on.
///
/// An index that indexes every path -- the standard JSON index -- always does. One that indexes a configured
/// subset answers an unindexed path with an empty bitmap, which is indistinguishable from "no document matches":
/// the query returns zero rows and nothing says why. Asking first turns that into a scan, which is slower and
/// right.
private boolean servesKey(JsonIndexReader jsonIndex) {
return jsonIndex.isPathIndexed(JSON_PATH_PREFIX + _keyName);
}

private FilterContext createFilterContext() {
Predicate rewritten = rewritePredicateForKey(_predicate);
if (rewritten == null) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -614,6 +614,8 @@ public void testSparseJsonIndexEqUsesPostings() {
postings.add(0);
postings.add(4);
when(jsonIndex.getMatchingDocIds(any(FilterContext.class))).thenReturn(postings);
// Mockito answers false for an unstubbed boolean, including an interface default that returns true.
when(jsonIndex.isPathIndexed(anyString())).thenReturn(true);

OpenStructDataSource osDs = withSparseJsonIndex(mockSparseSegmentSource(List.of("region"), Map.of()), jsonIndex);
MapFilterOperator op = new MapFilterOperator(mockSegment(osDs),
Expand All @@ -629,6 +631,8 @@ public void testSparseJsonIndexNotEqComplementsPostings() {
MutableRoaringBitmap postings = new MutableRoaringBitmap();
postings.add(0);
when(jsonIndex.getMatchingDocIds(any(FilterContext.class))).thenReturn(postings);
// Mockito answers false for an unstubbed boolean, including an interface default that returns true.
when(jsonIndex.isPathIndexed(anyString())).thenReturn(true);

OpenStructDataSource osDs = withSparseJsonIndex(mockSparseSegmentSource(List.of("region"), Map.of()), jsonIndex);
MapFilterOperator op = new MapFilterOperator(mockSegment(osDs),
Expand All @@ -638,6 +642,24 @@ public void testSparseJsonIndexNotEqComplementsPostings() {
assertEquals(countMatches(op), NUM_DOCS - 1);
}

/// An index that holds postings for only some paths answers an unindexed one with an empty bitmap, which reads
/// exactly like "nothing matched". Taking that answer would return zero rows and report no reason, so the key
/// has to fall back to the scan instead.
@Test
public void testSparseJsonIndexIsRefusedForAnUnindexedKey() {
JsonIndexReader jsonIndex = mock(JsonIndexReader.class);
when(jsonIndex.getMatchingDocIds(any(FilterContext.class))).thenReturn(new MutableRoaringBitmap());
when(jsonIndex.isPathIndexed(anyString())).thenReturn(false);

OpenStructDataSource osDs =
withSparseJsonIndex(mockSparseSegmentSource(List.of("region"), Map.of()), jsonIndex);
MapFilterOperator op = new MapFilterOperator(mockSegment(osDs),
makeEqPredicate(COLUMN, "region", "us"), mockQueryContext(), NUM_DOCS);

assertFalse(op.toExplainString().contains("delegateTo:json_match"),
"an index without postings for this key must not be asked");
}

@Test
public void testSparseJsonIndexRefusals() {
JsonIndexReader jsonIndex = mock(JsonIndexReader.class);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
*/
package org.apache.pinot.segment.local.segment.index.loader;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.annotations.VisibleForTesting;
import java.util.Collections;
import java.util.HashMap;
Expand Down Expand Up @@ -51,6 +52,7 @@
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.data.OpenStructNaming;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.ReadMode;
import org.apache.pinot.spi.utils.TimestampIndexUtils;

Expand Down Expand Up @@ -495,9 +497,6 @@ public IndexLoadingConfig withOpenStructChildConfigs(SegmentMetadataImpl segment
if (!childColumn.contains(OpenStructNaming.SEPARATOR) || indexConfigsByColName.containsKey(childColumn)) {
continue;
}
if (OpenStructNaming.isSparseColumn(childColumn)) {
continue;
}
String parentColumn = OpenStructNaming.parseParentColumn(childColumn);
FieldIndexConfigs parentConfigs = indexConfigsByColName.get(parentColumn);
if (parentConfigs == null) {
Expand All @@ -508,6 +507,16 @@ public IndexLoadingConfig withOpenStructChildConfigs(SegmentMetadataImpl segment
continue;
}
OpenStructIndexConfig openStructConfig = (OpenStructIndexConfig) osConfig;
if (OpenStructNaming.isSparseColumn(childColumn)) {
FieldIndexConfigs blobConfigs = sparseChildConfigs(openStructConfig, entry.getValue().getFieldSpec());
if (blobConfigs != null) {
if (updatedConfigs == null) {
updatedConfigs = new HashMap<>(indexConfigsByColName);
}
updatedConfigs.put(childColumn, blobConfigs);
}
continue;
}
String key = OpenStructNaming.parseKey(childColumn);
FieldConfig keyFieldConfig = openStructConfig.getValueFieldConfig(key);
if (keyFieldConfig == null) {
Expand Down Expand Up @@ -536,6 +545,36 @@ public void addOpenStructChildConfigs(SegmentMetadataImpl segmentMetadata) {
_resolvedIndexState = withOpenStructChildConfigs(segmentMetadata)._resolvedIndexState;
}

/// Index configs for the shared blob column, or null when the table asked for nothing on it.
///
/// The blob is an ordinary single-value STRING column holding each row's document as JSON, so an index that
/// reads JSON out of a STRING column applies to it unchanged -- a JSON index, or any other registered index
/// type named under `sparseFieldConfig.indexes`. What it must never get is a dictionary: the values are whole
/// documents, nearly all distinct, and a forward index of dict ids over them costs more than it saves. So the
/// encoding is forced RAW here rather than taken from the config.
@Nullable
private static FieldIndexConfigs sparseChildConfigs(OpenStructIndexConfig openStructConfig,
FieldSpec blobFieldSpec) {
FieldConfig sparseFieldConfig = openStructConfig.getSparseFieldConfig();
boolean legacyJsonIndex = openStructConfig.isSparseJsonIndex();
if (sparseFieldConfig == null && !legacyJsonIndex) {
return null;
}
FieldConfig.Builder builder = sparseFieldConfig != null
? new FieldConfig.Builder(sparseFieldConfig) : new FieldConfig.Builder(blobFieldSpec.getName());
builder.withEncodingType(FieldConfig.EncodingType.RAW);
if (legacyJsonIndex) {
// `sparseJsonIndex: true` predates sparseFieldConfig and means exactly one thing: a JSON index on the blob.
ObjectNode indexes = sparseFieldConfig != null && sparseFieldConfig.getIndexes() instanceof ObjectNode existing
? existing.deepCopy() : JsonUtils.newObjectNode();
if (!indexes.has(StandardIndexes.json().getPrettyName())) {
indexes.set(StandardIndexes.json().getPrettyName(), JsonUtils.newObjectNode());
}
builder.withIndexes(indexes);
}
return FieldIndexConfigsUtil.fromFieldConfig(builder.build(), blobFieldSpec);
}

public IndexLoadingConfig withKnownColumns(Set<String> columns) {
if (_knownColumns != null && _knownColumns.containsAll(columns)) {
return this;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import javax.annotation.Nullable;
Expand All @@ -33,6 +34,8 @@
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.datasource.DataSourceMetadata;
import org.apache.pinot.segment.spi.datasource.OpenStructDataSource;
import org.apache.pinot.segment.spi.index.IndexService;
import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.column.ColumnIndexContainer;
import org.apache.pinot.segment.spi.index.reader.ForwardIndexReader;
import org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
Expand All @@ -47,6 +50,10 @@
/// get virtual [SparseKeyDataSource]s backed by the shared blob parser; keys absent from the segment (no sparse blob,
/// or not listed in the sparse manifest) resolve to an all-null [NullDataSource].
public class ImmutableOpenStructDataSource extends BaseDataSource implements OpenStructDataSource {
/// Looked up by id rather than referenced: the composite JSON index ships as a plugin and is absent from most
/// deployments, so this resolves to nothing unless one registered it.
private static final String COMPOSITE_JSON_INDEX_ID = "composite_json_index";

private final ComplexFieldSpec _fieldSpec;
private final Map<String, DataSource> _perKeyDataSources;
@Nullable
Expand Down Expand Up @@ -189,7 +196,19 @@ public ColumnIndexContainer getIndexContainer(String key) {
@Override
@Nullable
public JsonIndexReader getSparseJsonIndex() {
return _sparseDataSource != null ? _sparseDataSource.getJsonIndex() : null;
if (_sparseDataSource == null) {
return null;
}
JsonIndexReader jsonIndex = _sparseDataSource.getJsonIndex();
if (jsonIndex != null) {
return jsonIndex;
}
// A composite JSON index reads as a [JsonIndexReader] and answers the same predicates, so a sparse key is
// served by whichever of the two the blob was given. This is the fallback an ordinary column already gets in
// [org.apache.pinot.core.operator.filter.MapFilterOperator]; without it here the index is built on the blob and
// then never consulted, and the key falls back to a full scan that silently costs what the index was for.
Optional<IndexType<?, ?, ?>> compositeIndex = IndexService.getInstance().getOptional(COMPOSITE_JSON_INDEX_ID);
return compositeIndex.map(indexType -> (JsonIndexReader) _sparseDataSource.getIndex(indexType)).orElse(null);
}

@Nullable
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -60,8 +60,9 @@ public class OpenStructIndexType
private static final List<String> EXTENSIONS = List.of(".open_struct.idx");
private static final String FIXED_FORWARD_INDEX =
"materialized keys always use a dictionary-encoded or LZ4 raw forward index";
private static final String KEY_ENCODING_RULE = "a key is dictionary-encoded when its encodingType is DICTIONARY or "
+ "its inverted index is enabled, and raw with LZ4 otherwise";
private static final String KEY_ENCODING_RULE = "a key is dictionary-encoded when its encodingType is "
+ "DICTIONARY, or indexes.dictionary enables one, or its inverted index is enabled, and raw with LZ4 "
+ "otherwise";

protected OpenStructIndexType() {
super(StandardIndexes.OPEN_STRUCT_ID);
Expand Down Expand Up @@ -175,7 +176,10 @@ private static void validatePerKeyFieldConfig(FieldConfig fieldConfig, String co
// Mirrors OpenStructColumnSplitter: a key gets a dictionary and a dictionary-encoded forward index when its
// encodingType is DICTIONARY or its inverted index is enabled. Inverted is the only vetted per-key index that
// requires a dictionary, so it overrides a RAW encodingType.
JsonNode declaredDictionary = indexes.get(StandardIndexes.DICTIONARY_ID);
boolean dictionaryEncoded = fieldConfig.getEncodingType() != FieldConfig.EncodingType.RAW
|| (declaredDictionary != null && declaredDictionary.isObject()
&& !declaredDictionary.path("disabled").asBoolean(false))
|| isEnabled(indexes.get(StandardIndexes.inverted().getPrettyName()));
if (forwardIndex != null && !forwardIndex.isNull()) {
validatePerKeyForwardIndex(forwardIndex, dictionaryEncoded, column, target);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -261,9 +261,6 @@ public Object[][] ignoredPerKeySettings() {
{"{'name': 'clicks', 'indexes': {'dictionary': {'disabled': true}}}",
"OPEN_STRUCT column 'payload': indexes.dictionary of key 'clicks' disables the dictionary, but the key is "
+ "built with one; "},
{"{'name': 'clicks', 'encodingType': 'RAW', 'indexes': {'dictionary': {}}}",
"OPEN_STRUCT column 'payload': indexes.dictionary of key 'clicks' enables the dictionary, but the key is "
+ "built without one; "},
// An enabled inverted index requires a dictionary, so a RAW key with one is still built with a dictionary.
{"{'name': 'clicks', 'encodingType': 'RAW', 'indexes': {'inverted': {}, 'dictionary': {'disabled': true}}}",
"OPEN_STRUCT column 'payload': indexes.dictionary of key 'clicks' disables the dictionary, but the key is "
Expand Down
Loading
Loading