This is an automated email from the ASF dual-hosted git repository.
raghavyadav01 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 88c3262d10a Support chained, forward-index-disabled, and
disabled-source derived columns (#19581)
88c3262d10a is described below
commit 88c3262d10ac94bb70aceded8771595d288b77ca
Author: RAGHVENDRA KUMAR YADAV <[email protected]>
AuthorDate: Tue Oct 20 15:33:51 2026 -0700
Support chained, forward-index-disabled, and disabled-source derived
columns (#19581)
* Support chained, forward-index-disabled, and disabled-source derived
columns
BaseDefaultColumnHandler carried three TODOs that limited derived columns
built
at segment load. All three are lifted here.
Chained derived columns. updateDefaultColumns iterated an unordered map and
resolved arguments against the segment metadata loaded before the run, so a
column created during the run was invisible. Columns are now visited in
dependency order (topological over the ADD actions), and the metadata
written
for each created column is recorded and consulted before the segment
metadata.
This also fixes a derived column reading a plain default column added in the
same reload, which failed the same way. A dependency cycle cannot be
satisfied
in any order: those columns are emitted last and take the default-value
path,
with the cycle logged rather than failing the segment load.
Source column with the forward index disabled. Previously an unconditional
UnsupportedOperationException. The forward index is now regenerated from the
dictionary and inverted index for the duration of the run and dropped
afterwards, mirroring the temporary-forward-index pattern BaseIndexHandler
already uses. When regeneration is not possible the column falls back to its
default value, or still fails when errorOnColumnBuildFailure is set.
Forward-index-disabled derived column. Previously skipped with a warning.
It is
now built with a forward index and a dictionary; ForwardIndexHandler runs
next,
queues DISABLE_FORWARD_INDEX and deletes the forward index in its
post-update
cleanup, once the inverted index has been built from it.
V3DefaultColumnHandler decided whether to transfer a forward or an inverted
index into the v3 file from the config. A forward-index-disabled multi-value
default column writes an inverted index but a derived one writes a forward
index, so the decision now follows the file that exists on disk.
DefaultColumnHandlerFactory gains a SegmentDirectory overload, needed by
InvertedIndexAndDictionaryBasedForwardIndexCreator; the existing signature
is
kept and delegates with null.
The transform configs are indexed by column and the parsed expressions
cached,
so each is parsed once per segment load rather than once for ordering and
again
for deriving. Reading back a created column's metadata is restricted to
columns
a transform function actually reads, so a schema evolution adding many plain
default columns does not re-parse metadata it will never consult. A table
with
no derived columns takes the same path as before: no expression is parsed
and
the action map is returned unchanged.
* Address review: chained value fidelity, metadata refresh, update
ordering, cycle fallback
Five correctness findings from the review of the previous commit.
Chained derived columns now evaluate on the expression results rather than
on
what the intermediate column stored. Ingestion evaluates a whole chain
before
converting anything to its stored type, so with a = 1, INT b = divide(a, 2)
and
DOUBLE c = plus(b, 1) it stores b = 0 and c = 1.5. Reading b back from its
index
gave c = 1.0 instead. The values a derived column produces are kept, before
conversion, for the columns another transform function reads, and dropped
at the
end of the run.
An argument's metadata is re-read after its forward index is regenerated.
The
rebuild rewrites the properties the reader works from -- the encoding flips
when
a raw index is restored, and TOTAL_NUMBER_OF_ENTRIES shrinks when a disabled
multi-value source held duplicates within a row -- so the copy taken
beforehand
computed the wrong offsets, the same reason BaseIndexHandler returns
refreshed
metadata from createForwardIndexIfNeeded.
Dependency ordering covers UPDATE actions, not just ADD. A source whose
default
value is changing is only correct once its own action has run; reading it
earlier
wrote the pre-update value into the chain permanently, and no later reload
put it
right.
Regenerating the forward index of a source created in the same run no longer
dereferences null. Such a column exists only in the metadata properties just
flushed, not in the segment directory's metadata, so the creator takes it
as an
argument; InvertedIndexAndDictionaryBasedForwardIndexCreator gains the
overload.
Every member of a dependency cycle now takes its default value. Returning on
re-entry left the enclosing calls free to emit the other members, so
whichever
one ran second read the first and derived a value from a default instead of
taking the documented fallback.
Five regression tests, each failing without the matching fix: the chain
compared
against a segment built through ingestion, the encoding flip during
regeneration,
the updated source ahead of a two-link chain, a derived column reading a
forward-index-disabled multi-value column added in the same reload, and the
cycle
asserted on actual default values rather than cardinality.
* Address review: null intermediates, refreshed source metadata, full cycle
detection
- Hand the raw expression results to the rest of the chain, captured before
nulls are replaced with the column's default, so a chain over a null
intermediate stores what ingestion stores.
- Cache the metadata refreshed after a forward index is regenerated for
every
source, not just ones created in the same run, so a second read of the
same
source works from it.
- Identify dependency cycles as strongly connected components, so every
member
of overlapping cycles falls back to its own default value.
* Fall back explicitly for a derived column reading a cycle member
A column reading a member of a transform dependency cycle relied on the
member's metadata being absent to take the default-value path. When the
cycle members already exist and are only being updated, their metadata is
present, so the dependent -- ordered before the postponed cycle -- read
their pre-update values and kept them for good.
Treat a cyclic argument as an explicit fallback when resolving a derived
column's arguments.
---
...IndexAndDictionaryBasedForwardIndexCreator.java | 17 +-
.../segment/index/loader/SegmentPreProcessor.java | 2 +-
.../defaultcolumn/BaseDefaultColumnHandler.java | 586 ++++++++++++++++++---
.../defaultcolumn/DefaultColumnHandlerFactory.java | 15 +-
.../defaultcolumn/V1DefaultColumnHandler.java | 7 +
.../defaultcolumn/V3DefaultColumnHandler.java | 26 +-
.../index/loader/SegmentPreProcessorTest.java | 509 +++++++++++++++++-
7 files changed, 1080 insertions(+), 82 deletions(-)
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/InvertedIndexAndDictionaryBasedForwardIndexCreator.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/InvertedIndexAndDictionaryBasedForwardIndexCreator.java
index fbbf894a9d7..cb6a735a7c8 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/InvertedIndexAndDictionaryBasedForwardIndexCreator.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/InvertedIndexAndDictionaryBasedForwardIndexCreator.java
@@ -24,6 +24,7 @@ import java.io.IOException;
import java.math.BigDecimal;
import java.util.HashMap;
import java.util.Map;
+import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
import
org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexType;
import
org.apache.pinot.segment.local.segment.index.forward.CompressionStatsMetadata;
@@ -119,6 +120,17 @@ public class
InvertedIndexAndDictionaryBasedForwardIndexCreator implements AutoC
SegmentDirectory.Writer segmentWriter, TableConfig tableConfig, String
columnName,
FieldIndexConfigs fieldIndexConfigs, boolean isTemporaryForwardIndex)
throws IOException {
+ this(segmentDirectory, segmentWriter, tableConfig, columnName,
fieldIndexConfigs, isTemporaryForwardIndex, null);
+ }
+
+ /// Same, for a column the segment directory's metadata does not list yet. A
column created during the current
+ /// segment load is only visible in `metadata.properties`, so its metadata
has to be handed in; passing `null`
+ /// keeps the usual lookup.
+ public InvertedIndexAndDictionaryBasedForwardIndexCreator(SegmentDirectory
segmentDirectory,
+ SegmentDirectory.Writer segmentWriter, TableConfig tableConfig, String
columnName,
+ FieldIndexConfigs fieldIndexConfigs, boolean isTemporaryForwardIndex,
+ @Nullable ColumnMetadata columnMetadata)
+ throws IOException {
_segmentDirectory = segmentDirectory;
_segmentWriter = segmentWriter;
_tableConfig = tableConfig;
@@ -128,7 +140,10 @@ public class
InvertedIndexAndDictionaryBasedForwardIndexCreator implements AutoC
_isTemporaryForwardIndex = isTemporaryForwardIndex;
_segmentMetadata = segmentDirectory.getSegmentMetadata();
- _columnMetadata = _segmentMetadata.getColumnMetadataFor(columnName);
+ _columnMetadata =
+ columnMetadata != null ? columnMetadata :
_segmentMetadata.getColumnMetadataFor(columnName);
+ Preconditions.checkState(_columnMetadata != null, "Found no metadata for
column: %s of segment: %s", columnName,
+ _segmentMetadata.getName());
_singleValue = _columnMetadata.isSingleValue();
_numDocs = _columnMetadata.getTotalDocs();
_cardinality = _columnMetadata.getCardinality();
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessor.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessor.java
index 6912e8f5833..a9da03a539d 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessor.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessor.java
@@ -167,7 +167,7 @@ public class SegmentPreProcessor implements AutoCloseable {
// Update default columns according to the schema.
DefaultColumnHandler defaultColumnHandler =
DefaultColumnHandlerFactory.getDefaultColumnHandler(indexDir,
segmentMetadata, _indexLoadingConfig,
- segmentWriter);
+ segmentWriter, _segmentDirectory);
defaultColumnHandler.updateDefaultColumns();
_segmentDirectory.reloadMetadata();
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java
index cd29c2045b0..1727d8c2ba6 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java
@@ -24,12 +24,20 @@ import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.math.BigDecimal;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Deque;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.function.IntFunction;
+import javax.annotation.Nullable;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.pinot.common.evaluator.FunctionEvaluatorFactory;
import org.apache.pinot.common.function.FunctionUtils;
@@ -52,6 +60,7 @@ import
org.apache.pinot.segment.local.segment.creator.impl.stats.StringColumnPre
import
org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexType;
import
org.apache.pinot.segment.local.segment.index.forward.CompressionStatsMetadata;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import
org.apache.pinot.segment.local.segment.index.loader.InvertedIndexAndDictionaryBasedForwardIndexCreator;
import org.apache.pinot.segment.local.segment.index.loader.LoaderUtils;
import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
import org.apache.pinot.segment.local.utils.ClusterConfigForTable;
@@ -71,6 +80,7 @@ import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.StandardIndexes;
import
org.apache.pinot.segment.spi.index.creator.DictionaryBasedInvertedIndexCreator;
import org.apache.pinot.segment.spi.index.creator.ForwardIndexCreator;
+import org.apache.pinot.segment.spi.index.metadata.ColumnMetadataImpl;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.segment.spi.index.reader.ForwardIndexReader;
import org.apache.pinot.segment.spi.store.SegmentDirectory;
@@ -138,13 +148,52 @@ public abstract class BaseDefaultColumnHandler implements
DefaultColumnHandler {
protected final TableConfig _tableConfig;
protected final Schema _schema;
protected final SegmentDirectory.Writer _segmentWriter;
+ /// Needed only to regenerate the forward index of a forward-index-disabled
source column of a derived column.
+ /// Null when the handler runs without one (the metadata-only
`needUpdateDefaultColumns` path), in which case such
+ /// a source cannot be read and the derived column falls back to its default
value.
+ @Nullable
+ protected final SegmentDirectory _segmentDirectory;
// NOTE: _segmentProperties shouldn't be used when checking whether default
column need to be created because at that
// time _segmentMetadata might not be loaded from a local file
private PropertiesConfiguration _segmentProperties;
+ /// Metadata of the columns created during this run. `_segmentMetadata` was
loaded before the run and never sees
+ /// them, so without this a derived column could not read a column added in
the same reload -- including another
+ /// derived column (chained derived columns).
+ private final Map<String, ColumnMetadata> _createdColumnMetadata = new
HashMap<>();
+
+ /// Values a derived column produced in this run, as the transform function
returned them -- before they were
+ /// converted to the column's stored type. Another column in the same chain
reads these rather than the index that
+ /// was just written, because ingestion evaluates the whole chain on the raw
expression results and converts only
+ /// at the end: with `a = 1`, `INT b = divide(a, 2)` stores 0, yet `DOUBLE c
= plus(b, 1)` must still see 0.5 and
+ /// store 1.5. Only kept for the columns a transform function actually
reads, and dropped at the end of the run.
+ private final Map<String, Object[]> _createdColumnValues = new HashMap<>();
+
+ /// Columns that sit in a transform-function dependency cycle. No order can
satisfy them, so each one is created
+ /// from its default value instead of its transform function -- every member
of the cycle, not only the first one
+ /// the traversal happens to reach.
+ private final Set<String> _cyclicColumns = new HashSet<>();
+
+ /// Source columns whose forward index was regenerated purely so a derived
column could read them. Dropped once
+ /// every default column has been created, restoring the
forward-index-disabled shape the config asks for.
+ private final Set<String> _tmpForwardIndexColumns = new LinkedHashSet<>();
+
+ /// Transform config per derived column, and the columns their expressions
read. Built once: the config list would
+ /// otherwise be scanned once per column being created, and the expressions
parsed twice - once to order the
+ /// columns, once to derive them. Null until the first lookup; empty when
the table declares no transform.
+ private Map<String, TransformConfig> _transformConfigsByColumn;
+ private Set<String> _derivedColumnArgumentNames;
+ private final Map<String, FunctionEvaluator> _functionEvaluators = new
HashMap<>();
+
protected BaseDefaultColumnHandler(File indexDir, SegmentMetadata
segmentMetadata,
IndexLoadingConfig indexLoadingConfig, SegmentDirectory.Writer
segmentWriter) {
+ this(indexDir, segmentMetadata, indexLoadingConfig, segmentWriter, null);
+ }
+
+ protected BaseDefaultColumnHandler(File indexDir, SegmentMetadata
segmentMetadata,
+ IndexLoadingConfig indexLoadingConfig, SegmentDirectory.Writer
segmentWriter,
+ @Nullable SegmentDirectory segmentDirectory) {
_indexDir = indexDir;
_segmentMetadata = segmentMetadata;
_indexLoadingConfig = indexLoadingConfig;
@@ -153,6 +202,7 @@ public abstract class BaseDefaultColumnHandler implements
DefaultColumnHandler {
_schema = _indexLoadingConfig.getSchema();
Preconditions.checkArgument(_schema != null, "Schema must be provided");
_segmentWriter = segmentWriter;
+ _segmentDirectory = segmentDirectory;
}
@Override
@@ -177,15 +227,22 @@ public abstract class BaseDefaultColumnHandler implements
DefaultColumnHandler {
return;
}
- // Update each default column based on the default column action.
+ // Update each default column based on the default column action. Columns
are visited in dependency order so a
+ // derived column is created after every column it reads that is being
created in this same run.
_segmentProperties =
SegmentMetadataUtils.getPropertiesConfiguration(_segmentMetadata);
- Iterator<Map.Entry<String, DefaultColumnAction>> entryIterator =
defaultColumnActionMap.entrySet().iterator();
- while (entryIterator.hasNext()) {
- Map.Entry<String, DefaultColumnAction> entry = entryIterator.next();
- // This method updates the metadata properties, need to save it later.
Remove the entry if the update failed.
- if (!updateDefaultColumn(entry.getKey(), entry.getValue())) {
- entryIterator.remove();
+ defaultColumnActionMap = orderByDependencies(defaultColumnActionMap);
+ try {
+ Iterator<Map.Entry<String, DefaultColumnAction>> entryIterator =
defaultColumnActionMap.entrySet().iterator();
+ while (entryIterator.hasNext()) {
+ Map.Entry<String, DefaultColumnAction> entry = entryIterator.next();
+ // This method updates the metadata properties, need to save it later.
Remove the entry if the update failed.
+ if (!updateDefaultColumn(entry.getKey(), entry.getValue())) {
+ entryIterator.remove();
+ }
}
+ } finally {
+ removeTemporaryForwardIndexes();
+ _createdColumnValues.clear();
}
// Update the segment metadata.
@@ -378,65 +435,99 @@ public abstract class BaseDefaultColumnHandler implements
DefaultColumnHandler {
/// Helper method to create the V1 indices (dictionary and forward index)
for a column, returns `true` if the
/// creation succeeds, `false` otherwise.
+ ///
+ /// A derived column whose argument is itself created in this same run is
supported: [#updateDefaultColumns] visits
+ /// the columns in dependency order and [#getColumnMetadata] sees the
metadata written a moment ago. A source
+ /// column with its forward index disabled is supported too, by regenerating
that index for the duration of the
+ /// run. An argument that is genuinely absent from the segment still falls
back to the default value.
protected boolean createColumnV1Indices(String column)
throws Exception {
boolean errorOnFailure = _indexLoadingConfig.isErrorOnColumnBuildFailure();
- IngestionConfig ingestionConfig = _tableConfig.getIngestionConfig();
- if (ingestionConfig != null && ingestionConfig.getTransformConfigs() !=
null) {
- List<TransformConfig> transformConfigs =
ingestionConfig.getTransformConfigs();
- for (TransformConfig transformConfig : transformConfigs) {
- if (transformConfig.getColumnName().equals(column)) {
- String transformFunction = transformConfig.getTransformFunction();
- FunctionEvaluator functionEvaluator =
FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction);
-
- // Check if all arguments exist in the segment
- // TODO: Support chained derived column
- List<String> arguments = functionEvaluator.getArguments();
- List<ColumnMetadata> argumentsMetadata = new
ArrayList<>(arguments.size());
- for (String argument : arguments) {
- ColumnMetadata columnMetadata =
_segmentMetadata.getColumnMetadataFor(argument);
- if (columnMetadata == null) {
- LOGGER.warn("Assigning default value to derived column: {}
because argument: {} does not exist in the "
- + "segment", column, argument);
- createDefaultValueColumnV1Indices(column);
- return true;
- }
- // TODO: Support creation of derived columns from forward index
disabled columns
- if (!_segmentWriter.hasIndexFor(argument,
StandardIndexes.forward())) {
- throw new UnsupportedOperationException(String.format("Operation
not supported! Cannot create a derived "
- + "column %s because argument: %s does not have a
forward index. Enable forward index and "
- + "refresh/backfill the segments to create a derived
column from source column", column,
- argument));
- }
- argumentsMetadata.add(columnMetadata);
- }
-
- // TODO: Support forward index disabled derived column
- if (isForwardIndexDisabled(column)) {
- LOGGER.warn("Skip creating forward index disabled derived column:
{}", column);
- if (errorOnFailure) {
- throw new UnsupportedOperationException(
- String.format("Failed to create forward index disabled
derived column: %s", column));
- }
- return false;
+ TransformConfig transformConfig = getTransformConfig(column);
+ if (transformConfig != null && _cyclicColumns.contains(column)) {
+ // No order satisfies a cycle, so every column in it takes the default
value rather than deriving from a
+ // sibling that was itself defaulted.
+ LOGGER.warn("Assigning default value to derived column: {} because it is
part of a transform function "
+ + "dependency cycle", column);
+ createDefaultValueColumnV1Indices(column);
+ recordCreatedColumnMetadata(column);
+ return true;
+ }
+ if (transformConfig != null) {
+ String transformFunction = transformConfig.getTransformFunction();
+ FunctionEvaluator functionEvaluator = getFunctionEvaluator(column,
transformFunction);
+
+ // Check if all arguments can be read from the segment
+ List<String> arguments = functionEvaluator.getArguments();
+ List<DerivedColumnArgument> argumentSources = new
ArrayList<>(arguments.size());
+ for (String argument : arguments) {
+ if (_cyclicColumns.contains(argument)) {
+ // A cycle member is created after everything that reads it and
takes its own default value. Reading it
+ // here would see either nothing or, when it already exists and is
only being updated, its pre-update
+ // values, baked into this column for good. Fall back explicitly
rather than rely on absent metadata.
+ LOGGER.warn("Assigning default value to derived column: {} because
argument: {} is part of a transform "
+ + "function dependency cycle", column, argument);
+ createDefaultValueColumnV1Indices(column);
+ recordCreatedColumnMetadata(column);
+ return true;
+ }
+ Object[] createdValues = _createdColumnValues.get(argument);
+ if (createdValues != null) {
+ // Derived earlier in this same run: use the expression results it
produced, not the converted values its
+ // index now holds, so the chain matches what ingestion would have
computed.
+ argumentSources.add(DerivedColumnArgument.ofValues(createdValues));
+ continue;
+ }
+ ColumnMetadata columnMetadata = getColumnMetadata(argument);
+ if (columnMetadata == null) {
+ LOGGER.warn("Assigning default value to derived column: {} because
argument: {} does not exist in the "
+ + "segment", column, argument);
+ createDefaultValueColumnV1Indices(column);
+ recordCreatedColumnMetadata(column);
+ return true;
+ }
+ // The values are read through the argument's forward index. When the
argument has it disabled, regenerate
+ // it from the dictionary and inverted index for the duration of this
run. Regeneration can change the
+ // metadata the reader needs, so take the refreshed copy it returns.
+ columnMetadata = materializeSourceForwardIndex(argument,
columnMetadata);
+ if (columnMetadata == null) {
+ // Regeneration needs the argument's dictionary and inverted index;
without them the only way to get the
+ // values back is a refresh or back-fill. This used to be an
unconditional failure, so keep failing when
+ // the table asks for it, and otherwise degrade the way a missing
argument already does.
+ if (errorOnFailure) {
+ throw new UnsupportedOperationException(String.format("Operation
not supported! Cannot create a derived "
+ + "column %s because argument: %s does not have a forward
index and it could not be regenerated "
+ + "from its dictionary and inverted index. Enable forward
index and refresh/backfill the segments "
+ + "to create a derived column from source column", column,
argument));
}
+ LOGGER.warn("Assigning default value to derived column: {} because
argument: {} has no forward index and "
+ + "it could not be regenerated", column, argument);
+ createDefaultValueColumnV1Indices(column);
+ recordCreatedColumnMetadata(column);
+ return true;
+ }
+ argumentSources.add(DerivedColumnArgument.ofColumn(columnMetadata));
+ }
- try {
- createDerivedColumnV1Indices(column, functionEvaluator,
argumentsMetadata, errorOnFailure);
- return true;
- } catch (Exception e) {
- LOGGER.error("Caught exception while creating derived column: {}
with transform function: {}", column,
- transformFunction, e);
- if (errorOnFailure) {
- throw e;
- }
- return false;
- }
+ try {
+ // A forward-index-disabled derived column is built with its forward
index like any other; ForwardIndexHandler
+ // queues DISABLE_FORWARD_INDEX for it right after this handler and
drops the index in its post-update
+ // cleanup, once every handler that needs it (the inverted index,
above all) has read it.
+ createDerivedColumnV1Indices(column, functionEvaluator,
argumentSources, errorOnFailure);
+ recordCreatedColumnMetadata(column);
+ return true;
+ } catch (Exception e) {
+ LOGGER.error("Caught exception while creating derived column: {} with
transform function: {}", column,
+ transformFunction, e);
+ if (errorOnFailure) {
+ throw e;
}
+ return false;
}
}
createDefaultValueColumnV1Indices(column);
+ recordCreatedColumnMetadata(column);
return true;
}
@@ -532,17 +623,23 @@ public abstract class BaseDefaultColumnHandler implements
DefaultColumnHandler {
}
/// Helper method to create the V1 indices (dictionary and forward index)
for a column with derived values.
- /// TODO:
- /// - Support chained derived column
- /// - Support forward index disabled derived column
private void createDerivedColumnV1Indices(String column, FunctionEvaluator
functionEvaluator,
- List<ColumnMetadata> argumentsMetadata, boolean errorOnFailure)
+ List<DerivedColumnArgument> argumentSources, boolean errorOnFailure)
throws Exception {
- // Initialize value readers for all arguments
- int numArguments = argumentsMetadata.size();
+ // Initialize value readers for the arguments that are read from the
segment. An argument derived earlier in this
+ // same run is served from the values it produced instead (see
_createdColumnValues).
+ int numArguments = argumentSources.size();
List<ValueReader> valueReaders = new ArrayList<>(numArguments);
- for (ColumnMetadata argumentMetadata : argumentsMetadata) {
- valueReaders.add(new ValueReader(argumentMetadata));
+ List<IntFunction<Object>> argumentValues = new ArrayList<>(numArguments);
+ for (DerivedColumnArgument argumentSource : argumentSources) {
+ Object[] values = argumentSource.getValues();
+ if (values != null) {
+ argumentValues.add(docId -> values[docId]);
+ } else {
+ ValueReader valueReader = new
ValueReader(argumentSource.getColumnMetadata());
+ valueReaders.add(valueReader);
+ argumentValues.add(valueReader::getValue);
+ }
}
FieldSpec fieldSpec = _schema.getFieldSpecFor(column);
@@ -561,10 +658,14 @@ public abstract class BaseDefaultColumnHandler implements
DefaultColumnHandler {
Object[] inputValues = new Object[numArguments];
int numDocs = _segmentMetadata.getTotalDocs();
Object[] outputValues = new Object[numDocs];
+ // The expression results as the rest of the chain has to see them:
nulls still null, values not yet converted
+ // to the stored type. Ingestion evaluates every expression before
either of those happens, so a column reading
+ // this one has to be handed this array rather than what the index ends
up holding. Null when nothing reads it.
+ Object[] expressionValues = newChainValues(column, numDocs);
PinotDataType outputValueType = null;
for (int i = 0; i < numDocs; i++) {
for (int j = 0; j < numArguments; j++) {
- inputValues[j] = valueReaders.get(j).getValue(i);
+ inputValues[j] = argumentValues.get(j).apply(i);
}
Object outputValue = null;
@@ -584,6 +685,10 @@ public abstract class BaseDefaultColumnHandler implements
DefaultColumnHandler {
}
}
+ if (expressionValues != null) {
+ expressionValues[i] = outputValue;
+ }
+
if (outputValue == null) {
outputValue = fieldSpec.getDefaultNullValue();
if (nullValueVectorCreator != null) {
@@ -611,9 +716,12 @@ public abstract class BaseDefaultColumnHandler implements
DefaultColumnHandler {
DictionaryIndexConfig dictionaryIndexConfig =
fieldIndexConfigs != null ?
fieldIndexConfigs.getConfig(StandardIndexes.dictionary())
: DictionaryIndexConfig.DEFAULT;
- boolean createDictionary = dictionaryIndexConfig.isEnabled();
+ // A forward-index-disabled column is served by its dictionary plus
inverted index, so it must be dictionary
+ // encoded whatever the dictionary config says -- the same requirement
Pinot enforces for non-derived columns.
+ boolean forwardIndexDisabled = isForwardIndexDisabled(column);
+ boolean createDictionary = dictionaryIndexConfig.isEnabled() ||
forwardIndexDisabled;
boolean useNoDictColumnStatsCollector = false;
- if (!dictionaryIndexConfig.isEnabled()) {
+ if (!createDictionary) {
useNoDictColumnStatsCollector =
ClusterConfigForTable.useOptimizedNoDictCollector(_tableConfig);
}
StatsCollectorConfig statsCollectorConfig = new
StatsCollectorConfig(_tableConfig, _schema, null);
@@ -761,6 +869,10 @@ public abstract class BaseDefaultColumnHandler implements
DefaultColumnHandler {
} else {
createDerivedColumnForwardIndexWithoutDictionary(column, fieldSpec,
outputValues, statsCollector);
}
+ // Only once the column exists: a column that failed to be created must
not feed the rest of the chain.
+ if (expressionValues != null) {
+ _createdColumnValues.put(column, expressionValues);
+ }
} finally {
for (ValueReader valueReader : valueReaders) {
valueReader.close();
@@ -1219,6 +1331,346 @@ public abstract class BaseDefaultColumnHandler
implements DefaultColumnHandler {
compressionMetadata.applyTo(_segmentProperties, column);
}
+ /// Reorders the actions so that a column is created after every column it
reads that is also being created in
+ /// this run -- the dependency that makes chained derived columns work, and
that also lets a derived column read a
+ /// plain default column added in the same reload. Non-derived columns keep
their original relative order.
+ ///
+ /// A dependency cycle cannot be satisfied in any order, so the columns in
it are emitted last, unordered: each
+ /// still gets created, falling back to its default value because its
argument is not in the segment yet. The
+ /// cycle is logged rather than failing the segment load.
+ private Map<String, DefaultColumnAction> orderByDependencies(Map<String,
DefaultColumnAction> actionMap) {
+ Map<String, List<String>> dependencies = new HashMap<>();
+ for (Map.Entry<String, DefaultColumnAction> entry : actionMap.entrySet()) {
+ if (!entry.getValue().isRemoveAction()) {
+ List<String> arguments = derivedColumnArguments(entry.getKey());
+ if (arguments != null) {
+ List<String> pending = new ArrayList<>(arguments.size());
+ for (String argument : arguments) {
+ // Only columns written in this run constrain the order; anything
already in its final shape is readable.
+ // An updated argument counts as much as an added one: its values
are only correct once its own action has
+ // run, so reading it earlier would bake the pre-update values
into the derived column for good.
+ DefaultColumnAction argumentAction = actionMap.get(argument);
+ if (argumentAction != null && !argumentAction.isRemoveAction() &&
!argument.equals(entry.getKey())) {
+ pending.add(argument);
+ }
+ }
+ if (!pending.isEmpty()) {
+ dependencies.put(entry.getKey(), pending);
+ }
+ }
+ }
+ }
+ if (dependencies.isEmpty()) {
+ return actionMap;
+ }
+
+ markCyclicColumns(dependencies);
+
+ Map<String, DefaultColumnAction> ordered = new
LinkedHashMap<>(actionMap.size());
+ for (String column : actionMap.keySet()) {
+ visitForOrdering(column, actionMap, dependencies, ordered);
+ }
+ // Columns dropped by a cycle still have to be created; they land last and
take the default-value path.
+ for (Map.Entry<String, DefaultColumnAction> entry : actionMap.entrySet()) {
+ ordered.putIfAbsent(entry.getKey(), entry.getValue());
+ }
+ return ordered;
+ }
+
+ private void visitForOrdering(String column, Map<String,
DefaultColumnAction> actionMap,
+ Map<String, List<String>> dependencies, Map<String, DefaultColumnAction>
ordered) {
+ // A column on a cycle is left out here and appended at the end, so that
anything reading it is created first and
+ // takes the fallback for a cyclic argument (see [#createColumnV1Indices])
instead of deriving a value from a
+ // sibling's default. Skipping those columns is also what bounds this
traversal: every edge that closes a cycle
+ // leads into one of them.
+ if (ordered.containsKey(column) || _cyclicColumns.contains(column) ||
!actionMap.containsKey(column)) {
+ return;
+ }
+ for (String dependency : dependencies.getOrDefault(column, List.of())) {
+ visitForOrdering(dependency, actionMap, dependencies, ordered);
+ }
+ // A dependency stuck in a cycle is absent from `ordered`; emitting this
column anyway is safe because the cyclic
+ // argument makes it take the default-value path.
+ ordered.put(column, actionMap.get(column));
+ }
+
+ /// Records every column that sits on a dependency cycle, so that each of
them takes its own default value instead
+ /// of deriving from a sibling that was itself defaulted.
+ ///
+ /// Cycle membership is a strongly connected component of the dependency
graph, not just the path on which a
+ /// depth-first search happens to find a back edge. With `a = plus(b, c)`,
`b = plus(a, 1)` and `c = plus(b, 1)`
+ /// the first back edge only closes over `a` and `b`, yet `c` reaches itself
through `b` and `a` just the same, so
+ /// all three are cyclic. Tarjan's algorithm names the whole component in
one pass.
+ private void markCyclicColumns(Map<String, List<String>> dependencies) {
+ SccState state = new SccState();
+ for (String column : dependencies.keySet()) {
+ if (!state._index.containsKey(column)) {
+ strongConnect(column, dependencies, state);
+ }
+ }
+ }
+
+ private void strongConnect(String column, Map<String, List<String>>
dependencies, SccState state) {
+ int index = state._nextIndex++;
+ state._index.put(column, index);
+ state._lowLink.put(column, index);
+ state._stack.push(column);
+ state._onStack.add(column);
+
+ for (String dependency : dependencies.getOrDefault(column, List.of())) {
+ Integer dependencyIndex = state._index.get(dependency);
+ if (dependencyIndex == null) {
+ strongConnect(dependency, dependencies, state);
+ state._lowLink.put(column, Math.min(state._lowLink.get(column),
state._lowLink.get(dependency)));
+ } else if (state._onStack.contains(dependency)) {
+ state._lowLink.put(column, Math.min(state._lowLink.get(column),
dependencyIndex));
+ }
+ }
+
+ if (state._lowLink.get(column) == index) {
+ List<String> component = new ArrayList<>();
+ String member;
+ do {
+ member = state._stack.pop();
+ state._onStack.remove(member);
+ component.add(member);
+ } while (!member.equals(column));
+ // A component of one is a plain column: a self-reference never gets an
edge in the first place, since a
+ // transform reading the column it produces is reading the pre-reload
values and is left alone.
+ if (component.size() > 1) {
+ LOGGER.warn("Derived columns form a dependency cycle: {}. They will be
assigned default values instead of "
+ + "being derived; break the cycle in the transform configs.",
component);
+ _cyclicColumns.addAll(component);
+ }
+ }
+ }
+
+ /// The bookkeeping of one run of Tarjan's algorithm over the dependency
graph (see [#markCyclicColumns]).
+ private static class SccState {
+ final Map<String, Integer> _index = new HashMap<>();
+ final Map<String, Integer> _lowLink = new HashMap<>();
+ final Deque<String> _stack = new ArrayDeque<>();
+ final Set<String> _onStack = new HashSet<>();
+ int _nextIndex;
+ }
+
+ /// The arguments of the transform function that derives this column, or
null when the column is not derived.
+ @Nullable
+ private List<String> derivedColumnArguments(String column) {
+ TransformConfig transformConfig = getTransformConfig(column);
+ if (transformConfig == null) {
+ return null;
+ }
+ try {
+ return getFunctionEvaluator(column,
transformConfig.getTransformFunction()).getArguments();
+ } catch (Exception e) {
+ // An unparseable transform function fails later in
createColumnV1Indices with the full context; ordering just
+ // treats the column as dependency-free.
+ return null;
+ }
+ }
+
+ @Nullable
+ private TransformConfig getTransformConfig(String column) {
+ if (_transformConfigsByColumn == null) {
+ buildTransformConfigIndex();
+ }
+ return _transformConfigsByColumn.get(column);
+ }
+
+ /// Indexes the transform configs by the column they produce, and collects
every column their expressions read.
+ /// An unparseable expression is left out of the argument set; it fails
later in [#createColumnV1Indices] with the
+ /// transform function in the message.
+ private void buildTransformConfigIndex() {
+ _transformConfigsByColumn = new HashMap<>();
+ _derivedColumnArgumentNames = new HashSet<>();
+ IngestionConfig ingestionConfig = _tableConfig.getIngestionConfig();
+ List<TransformConfig> transformConfigs =
+ ingestionConfig != null ? ingestionConfig.getTransformConfigs() : null;
+ if (transformConfigs == null) {
+ return;
+ }
+ for (TransformConfig transformConfig : transformConfigs) {
+ _transformConfigsByColumn.putIfAbsent(transformConfig.getColumnName(),
transformConfig);
+ try {
+ _derivedColumnArgumentNames.addAll(
+ getFunctionEvaluator(transformConfig.getColumnName(),
transformConfig.getTransformFunction())
+ .getArguments());
+ } catch (Exception e) {
+ LOGGER.debug("Could not parse the transform function of column: {}
while indexing the transform configs",
+ transformConfig.getColumnName(), e);
+ }
+ }
+ }
+
+ /// The parsed expression for a derived column. Cached because ordering the
columns and deriving them both need it.
+ /// A parse failure propagates and is not cached, so the caller still sees
it.
+ private FunctionEvaluator getFunctionEvaluator(String column, String
transformFunction) {
+ return _functionEvaluators.computeIfAbsent(column,
+ k ->
FunctionEvaluatorFactory.getExpressionEvaluator(transformFunction));
+ }
+
+ /// Metadata for a column, including one created earlier in this same run
(see [#_createdColumnMetadata]).
+ @Nullable
+ private ColumnMetadata getColumnMetadata(String column) {
+ ColumnMetadata created = _createdColumnMetadata.get(column);
+ return created != null ? created :
_segmentMetadata.getColumnMetadataFor(column);
+ }
+
+ /// Records the metadata just written for a created column so later columns
in this run can read it. Only columns
+ /// that a transform function actually reads are read back: a schema
evolution adding hundreds of plain default
+ /// columns would otherwise re-parse the metadata of every one of them for
nothing.
+ private void recordCreatedColumnMetadata(String column) {
+ if (_transformConfigsByColumn == null) {
+ buildTransformConfigIndex();
+ }
+ if (!_derivedColumnArgumentNames.contains(column)) {
+ return;
+ }
+ try {
+ _createdColumnMetadata.put(column,
+ ColumnMetadataImpl.fromPropertiesConfiguration(_segmentProperties,
_segmentMetadata.getTotalDocs(), column));
+ } catch (Exception e) {
+ // Only chained reads need this; a failure here must not fail the column
that was created successfully.
+ LOGGER.warn("Could not read back the metadata of newly created column:
{}; a derived column reading it in this "
+ + "same reload will fall back to its default value", column, e);
+ }
+ }
+
+ /// An array to collect the raw expression results of a derived column in,
to hand to a column later in the chain
+ /// (see [#_createdColumnValues]), or null when no transform function reads
this column and nothing would ever look
+ /// at them.
+ @Nullable
+ private Object[] newChainValues(String column, int numDocs) {
+ if (_transformConfigsByColumn == null) {
+ buildTransformConfigIndex();
+ }
+ return _derivedColumnArgumentNames.contains(column) ? new Object[numDocs]
: null;
+ }
+
+ /// Makes a forward index available for a source column that has it
disabled, by regenerating it from the
+ /// dictionary and inverted index. The regenerated index is temporary:
[#removeTemporaryForwardIndexes] drops it
+ /// once every default column is created, so the column keeps the shape its
config asks for.
+ ///
+ /// Returns the metadata to read the column with -- `columnMetadata` when
the forward index was already there, and
+ /// a refreshed copy when it was rebuilt, since the rebuild can change the
encoding and the entry count the reader
+ /// derives its offsets from.
+ ///
+ /// Returns null when regeneration is not possible -- no segment directory
(the metadata-only path), or the
+ /// dictionary / inverted index the rebuild needs is missing. The caller
then falls back to the default value
+ /// rather than failing the load.
+ @Nullable
+ private ColumnMetadata materializeSourceForwardIndex(String column,
ColumnMetadata columnMetadata) {
+ if (_segmentWriter.hasIndexFor(column, StandardIndexes.forward())) {
+ return columnMetadata;
+ }
+ if (_segmentDirectory == null) {
+ return null;
+ }
+ if (!_segmentWriter.hasIndexFor(column, StandardIndexes.dictionary())
+ || !_segmentWriter.hasIndexFor(column, StandardIndexes.inverted())) {
+ LOGGER.warn("Cannot regenerate the forward index of column: {} to derive
from it: it needs both a dictionary "
+ + "({}) and an inverted index ({})", column,
+ _segmentWriter.hasIndexFor(column, StandardIndexes.dictionary()) ?
"present" : "missing",
+ _segmentWriter.hasIndexFor(column, StandardIndexes.inverted()) ?
"present" : "missing");
+ return null;
+ }
+ FieldIndexConfigs fieldIndexConfigs =
_indexLoadingConfig.getFieldIndexConfig(column);
+ if (fieldIndexConfigs == null) {
+ return null;
+ }
+ try {
+ LOGGER.info("Temporarily regenerating the forward index of
forward-index-disabled column: {} so a derived "
+ + "column can read it", column);
+ // The creator writes metadata.properties itself
(SegmentMetadataUtils#updateMetadataProperties). Our own
+ // in-memory copy was loaded before this point and is saved at the end
of the run, so flush the columns
+ // created so far to disk and re-read afterwards -- otherwise the save
would discard the creator's writes.
+ if (_segmentProperties != null) {
+ SegmentMetadataUtils.savePropertiesConfiguration(_segmentProperties,
_segmentMetadata.getIndexDir());
+ }
+ // A column created earlier in this run is not in the segment
directory's metadata -- it only exists in the
+ // properties just flushed -- so hand the creator the metadata we
already hold instead of letting it look up
+ // one that is not there.
+ new
InvertedIndexAndDictionaryBasedForwardIndexCreator(_segmentDirectory,
_segmentWriter, _tableConfig, column,
+ fieldIndexConfigs, true, columnMetadata).regenerateForwardIndex();
+ if (_segmentProperties != null) {
+ _segmentProperties =
SegmentMetadataUtils.getPropertiesConfiguration(_segmentMetadata);
+ }
+ } catch (Exception e) {
+ LOGGER.warn("Failed to regenerate the forward index of column: {}",
column, e);
+ return null;
+ }
+ if (!_segmentWriter.hasIndexFor(column, StandardIndexes.forward())) {
+ return null;
+ }
+ _tmpForwardIndexColumns.add(column);
+ return refreshColumnMetadata(column, columnMetadata);
+ }
+
+ /// Re-reads a column's metadata after its forward index was regenerated.
The rebuild rewrites the properties the
+ /// readers work from -- the encoding flips when a raw index is restored,
and `TOTAL_NUMBER_OF_ENTRIES` shrinks
+ /// when a disabled multi-value source held duplicates within a row -- so
the copy taken beforehand would compute
+ /// the wrong offsets. Falls back to that copy when the re-read fails, which
leaves the previous behaviour.
+ private ColumnMetadata refreshColumnMetadata(String column, ColumnMetadata
previous) {
+ if (_segmentProperties == null) {
+ return previous;
+ }
+ try {
+ ColumnMetadata refreshed =
+ ColumnMetadataImpl.fromPropertiesConfiguration(_segmentProperties,
_segmentMetadata.getTotalDocs(), column);
+ // Every later read of this column has to get the refreshed copy, not
just the caller holding it now: a
+ // transform as plain as `plus(source, source)` reads the column twice,
and the second read finds the
+ // regenerated forward index already in place and so never comes back
through here.
+ _createdColumnMetadata.put(column, refreshed);
+ return refreshed;
+ } catch (Exception e) {
+ LOGGER.warn("Could not re-read the metadata of column: {} after
regenerating its forward index", column, e);
+ return previous;
+ }
+ }
+
+ /// Drops the forward indexes regenerated by
[#materializeSourceForwardIndex]. Mirrors
+ /// `BaseIndexHandler#postUpdateIndicesCleanup`, which does the same for the
index handlers.
+ private void removeTemporaryForwardIndexes() {
+ for (String column : _tmpForwardIndexColumns) {
+ try {
+ _segmentWriter.removeIndex(column, StandardIndexes.forward());
+ } catch (Exception e) {
+ LOGGER.warn("Could not remove the temporary forward index of column:
{}", column, e);
+ }
+ }
+ _tmpForwardIndexColumns.clear();
+ }
+
+ /// Where one argument of a transform function gets its values from: a
column in the segment, read through its
+ /// forward index, or the results a derived column produced earlier in this
same run.
+ private static class DerivedColumnArgument {
+ private final ColumnMetadata _columnMetadata;
+ private final Object[] _values;
+
+ private DerivedColumnArgument(@Nullable ColumnMetadata columnMetadata,
@Nullable Object[] values) {
+ _columnMetadata = columnMetadata;
+ _values = values;
+ }
+
+ static DerivedColumnArgument ofColumn(ColumnMetadata columnMetadata) {
+ return new DerivedColumnArgument(columnMetadata, null);
+ }
+
+ static DerivedColumnArgument ofValues(Object[] values) {
+ return new DerivedColumnArgument(null, values);
+ }
+
+ ColumnMetadata getColumnMetadata() {
+ return _columnMetadata;
+ }
+
+ @Nullable
+ Object[] getValues() {
+ return _values;
+ }
+ }
+
@SuppressWarnings("rawtypes")
private class ValueReader implements Closeable {
final ForwardIndexReader _forwardIndexReader;
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/DefaultColumnHandlerFactory.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/DefaultColumnHandlerFactory.java
index 5058ac91e80..778048dd7e6 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/DefaultColumnHandlerFactory.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/DefaultColumnHandlerFactory.java
@@ -19,6 +19,7 @@
package org.apache.pinot.segment.local.segment.index.loader.defaultcolumn;
import java.io.File;
+import javax.annotation.Nullable;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import org.apache.pinot.segment.spi.creator.SegmentVersion;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
@@ -31,10 +32,20 @@ public class DefaultColumnHandlerFactory {
public static DefaultColumnHandler getDefaultColumnHandler(File indexDir,
SegmentMetadataImpl segmentMetadata,
IndexLoadingConfig indexLoadingConfig, SegmentDirectory.Writer
segmentWriter) {
+ return getDefaultColumnHandler(indexDir, segmentMetadata,
indexLoadingConfig, segmentWriter, null);
+ }
+
+ /// `segmentDirectory` lets the handler regenerate the forward index of a
forward-index-disabled source column so a
+ /// derived column can read it; pass null when there is none to work with.
+ public static DefaultColumnHandler getDefaultColumnHandler(File indexDir,
SegmentMetadataImpl segmentMetadata,
+ IndexLoadingConfig indexLoadingConfig, SegmentDirectory.Writer
segmentWriter,
+ @Nullable SegmentDirectory segmentDirectory) {
if (segmentMetadata.getVersion() == SegmentVersion.v3) {
- return new V3DefaultColumnHandler(indexDir, segmentMetadata,
indexLoadingConfig, segmentWriter);
+ return new V3DefaultColumnHandler(indexDir, segmentMetadata,
indexLoadingConfig, segmentWriter,
+ segmentDirectory);
} else {
- return new V1DefaultColumnHandler(indexDir, segmentMetadata,
indexLoadingConfig, segmentWriter);
+ return new V1DefaultColumnHandler(indexDir, segmentMetadata,
indexLoadingConfig, segmentWriter,
+ segmentDirectory);
}
}
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/V1DefaultColumnHandler.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/V1DefaultColumnHandler.java
index 1a2935c4c6b..56cac3a6436 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/V1DefaultColumnHandler.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/V1DefaultColumnHandler.java
@@ -19,6 +19,7 @@
package org.apache.pinot.segment.local.segment.index.loader.defaultcolumn;
import java.io.File;
+import javax.annotation.Nullable;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.segment.spi.store.SegmentDirectory;
@@ -34,6 +35,12 @@ public class V1DefaultColumnHandler extends
BaseDefaultColumnHandler {
super(indexDir, segmentMetadata, indexLoadingConfig, segmentWriter);
}
+ public V1DefaultColumnHandler(File indexDir, SegmentMetadataImpl
segmentMetadata,
+ IndexLoadingConfig indexLoadingConfig, SegmentDirectory.Writer
segmentWriter,
+ @Nullable SegmentDirectory segmentDirectory) {
+ super(indexDir, segmentMetadata, indexLoadingConfig, segmentWriter,
segmentDirectory);
+ }
+
@Override
protected boolean updateDefaultColumn(String column, DefaultColumnAction
action)
throws Exception {
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/V3DefaultColumnHandler.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/V3DefaultColumnHandler.java
index c4ac70bad5d..1d3a1cca772 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/V3DefaultColumnHandler.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/V3DefaultColumnHandler.java
@@ -20,6 +20,7 @@ package
org.apache.pinot.segment.local.segment.index.loader.defaultcolumn;
import com.google.common.base.Preconditions;
import java.io.File;
+import javax.annotation.Nullable;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import org.apache.pinot.segment.local.segment.index.loader.LoaderUtils;
import org.apache.pinot.segment.spi.V1Constants;
@@ -39,6 +40,12 @@ public class V3DefaultColumnHandler extends
BaseDefaultColumnHandler {
super(indexDir, segmentMetadata, indexLoadingConfig, segmentWriter);
}
+ public V3DefaultColumnHandler(File indexDir, SegmentMetadataImpl
segmentMetadata,
+ IndexLoadingConfig indexLoadingConfig, SegmentDirectory.Writer
segmentWriter,
+ @Nullable SegmentDirectory segmentDirectory) {
+ super(indexDir, segmentMetadata, indexLoadingConfig, segmentWriter,
segmentDirectory);
+ }
+
@Override
protected boolean updateDefaultColumn(String column, DefaultColumnAction
action)
throws Exception {
@@ -61,7 +68,6 @@ public class V3DefaultColumnHandler extends
BaseDefaultColumnHandler {
FieldSpec fieldSpec = _schema.getFieldSpecFor(column);
Preconditions.checkNotNull(fieldSpec);
boolean isSingleValue = fieldSpec.isSingleValueField();
- boolean forwardIndexDisabled = !isSingleValue &&
isForwardIndexDisabled(column);
File forwardIndexFile = null;
File invertedIndexFile = null;
@@ -74,15 +80,17 @@ public class V3DefaultColumnHandler extends
BaseDefaultColumnHandler {
forwardIndexFile = new File(_indexDir, column +
V1Constants.Indexes.RAW_SV_FORWARD_INDEX_FILE_EXTENSION);
}
} else {
- if (forwardIndexDisabled) {
- // An inverted index is created instead of forward index for
multi-value columns with forward index disabled
- // Note that we don't currently support creation of forward index
disabled derived columns
+ forwardIndexFile = new File(_indexDir, column +
V1Constants.Indexes.UNSORTED_MV_FORWARD_INDEX_FILE_EXTENSION);
+ if (!forwardIndexFile.exists()) {
+ forwardIndexFile = new File(_indexDir, column +
V1Constants.Indexes.RAW_MV_FORWARD_INDEX_FILE_EXTENSION);
+ }
+ if (!forwardIndexFile.exists() && isForwardIndexDisabled(column)) {
+ // A forward-index-disabled multi-value DEFAULT column is written as
an inverted index instead of a forward
+ // index (see
BaseDefaultColumnHandler#createDefaultValueColumnV1Indices). A DERIVED column
always writes a
+ // forward index -- ForwardIndexHandler drops it in its post-update
cleanup once the inverted index has been
+ // built from it -- so the on-disk file, not the config, decides which
one to transfer here.
+ forwardIndexFile = null;
invertedIndexFile = new File(_indexDir, column +
V1Constants.Indexes.BITMAP_INVERTED_INDEX_FILE_EXTENSION);
- } else {
- forwardIndexFile = new File(_indexDir, column +
V1Constants.Indexes.UNSORTED_MV_FORWARD_INDEX_FILE_EXTENSION);
- if (!forwardIndexFile.exists()) {
- forwardIndexFile = new File(_indexDir, column +
V1Constants.Indexes.RAW_MV_FORWARD_INDEX_FILE_EXTENSION);
- }
}
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessorTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessorTest.java
index 12105b176e6..ccc323e742c 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessorTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessorTest.java
@@ -46,6 +46,7 @@ import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationD
import
org.apache.pinot.segment.local.segment.index.converter.SegmentV1V2ToV3FormatConverter;
import
org.apache.pinot.segment.local.segment.index.loader.columnminmaxvalue.ColumnMinMaxValueGeneratorMode;
import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.segment.readers.PinotSegmentRecordReader;
import org.apache.pinot.segment.local.segment.store.SegmentLocalFSDirectory;
import org.apache.pinot.segment.local.startree.StarTreeBuilderUtils;
import org.apache.pinot.segment.local.utils.SegmentOperationsThrottler;
@@ -80,8 +81,10 @@ import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
+import org.apache.pinot.spi.data.DimensionFieldSpec;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.MetricFieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.utils.ByteArray;
@@ -185,6 +188,7 @@ public class SegmentPreProcessorTest implements
PinotBuffersAfterClassCheckRule
private List<StarTreeIndexConfig> _starTreeIndexConfigs;
private boolean _enableDefaultStarTree;
private boolean _compressionStatsEnabled;
+ private boolean _nullHandlingEnabled;
public SegmentPreProcessorTest()
throws IOException {
@@ -258,6 +262,7 @@ public class SegmentPreProcessorTest implements
PinotBuffersAfterClassCheckRule
_starTreeIndexConfigs = null;
_enableDefaultStarTree = false;
_compressionStatsEnabled = false;
+ _nullHandlingEnabled = true;
}
@AfterMethod
@@ -274,13 +279,28 @@ public class SegmentPreProcessorTest implements
PinotBuffersAfterClassCheckRule
}
}
+ /// Builds a segment whose own generation applies the schema given --
including any transform config in
+ /// `_ingestionConfig`, which is how a derived column is produced on the
ingestion path rather than at load.
+ private void buildSegment(SegmentVersion segmentVersion, Schema schema)
+ throws Exception {
+ buildV1Segment(schema);
+ if (segmentVersion == SegmentVersion.v3) {
+ convertV1SegmentToV3();
+ }
+ }
+
private void buildV1Segment()
throws Exception {
+ buildV1Segment(_schema);
+ }
+
+ private void buildV1Segment(Schema schema)
+ throws Exception {
FileUtils.deleteQuietly(TEMP_DIR);
SegmentGeneratorConfig config =
SegmentTestUtils.getSegmentGeneratorConfigWithSchema(_avroFile,
TEMP_DIR, RAW_TABLE_NAME, createTableConfig(),
- _schema);
+ schema);
config.setInstanceType(InstanceType.SERVER);
config.setOutDir(TEMP_DIR.getPath());
config.setSegmentName(SEGMENT_NAME);
@@ -307,7 +327,7 @@ public class SegmentPreProcessorTest implements
PinotBuffersAfterClassCheckRule
.setInvertedIndexColumns(new ArrayList<>(_invertedIndexColumns))
.setRangeIndexColumns(new ArrayList<>(_rangeIndexColumns))
.setFieldConfigList(new ArrayList<>(_fieldConfigMap.values()))
- .setNullHandlingEnabled(true)
+ .setNullHandlingEnabled(_nullHandlingEnabled)
.setOptimizeNoDictStatsCollection(true)
.setIngestionConfig(_ingestionConfig)
.build();
@@ -351,6 +371,19 @@ public class SegmentPreProcessorTest implements
PinotBuffersAfterClassCheckRule
/// Cartesian product of segment version (v1/v3) and geo-column encoding
(DICTIONARY/RAW), so the H3 reload test
/// exercises both
[org.apache.pinot.segment.local.segment.index.loader.invertedindex.H3IndexHandler]
/// paths: the dictionary path and the raw
`forwardIndexReader.getBytes(...)` path.
+ /// Cartesian product of segment version (v1/v3) and the table's null
handling flag, for the derived-column tests
+ /// that have to behave the same either way: a transform expression yields
null regardless of the flag, which only
+ /// decides whether a null vector is written alongside the column.
+ @DataProvider(name = "versionAndNullHandling")
+ public Object[][] versionAndNullHandling() {
+ return new Object[][]{
+ {SegmentVersion.v1, true},
+ {SegmentVersion.v3, true},
+ {SegmentVersion.v1, false},
+ {SegmentVersion.v3, false}
+ };
+ }
+
@DataProvider(name = "h3VersionAndEncoding")
public Object[][] h3VersionAndEncoding() {
return new Object[][]{
@@ -2639,4 +2672,476 @@ public class SegmentPreProcessorTest implements
PinotBuffersAfterClassCheckRule
validateIndexDoesNotExist(StandardIndexes.inverted(),
NEWLY_ADDED_FORWARD_INDEX_DISABLED_COL_MV);
validateIndexExists(StandardIndexes.dictionary(),
NEWLY_ADDED_FORWARD_INDEX_DISABLED_COL_MV);
}
+
+ /// A derived column whose argument is another derived column added in the
same reload. The transform configs are
+ /// listed dependent-first on purpose: the handler has to order them by
dependency, not by config order.
+ @Test(dataProvider = "bothV1AndV3")
+ public void testChainedDerivedColumns(SegmentVersion segmentVersion)
+ throws Exception {
+ buildSegment(segmentVersion);
+
+ _ingestionConfig.setTransformConfigs(
+ List.of(new TransformConfig(NEW_INT_METRIC_COLUMN_NAME,
"plus(newIntSVDimension, 1)"),
+ new TransformConfig(NEW_INT_SV_DIMENSION_COLUMN_NAME,
"plus(column1, 1)")));
+ runPreProcessor(_newColumnsSchema1);
+
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(INDEX_DIR);
+ ColumnMetadata sourceMetadata =
segmentMetadata.getColumnMetadataFor(COLUMN1_NAME);
+ int sourceMin = (Integer) sourceMetadata.getMinValue();
+ int sourceMax = (Integer) sourceMetadata.getMaxValue();
+
+ ColumnMetadata firstLink =
segmentMetadata.getColumnMetadataFor(NEW_INT_SV_DIMENSION_COLUMN_NAME);
+ assertTrue(firstLink.isAutoGenerated());
+ assertEquals(firstLink.getMinValue(), sourceMin + 1);
+ assertEquals(firstLink.getMaxValue(), sourceMax + 1);
+
+ // The chained column is derived from the column above, not filled with
its default value.
+ ColumnMetadata secondLink =
segmentMetadata.getColumnMetadataFor(NEW_INT_METRIC_COLUMN_NAME);
+ assertTrue(secondLink.isAutoGenerated());
+ assertEquals(secondLink.getMinValue(), sourceMin + 2);
+ assertEquals(secondLink.getMaxValue(), sourceMax + 2);
+ assertEquals(secondLink.getCardinality(), sourceMetadata.getCardinality());
+ }
+
+ /// A derived column whose argument is a plain default column added in the
same reload. Before dependency
+ /// ordering this hit the same bug as chaining: the argument was not yet in
the segment metadata, so the derived
+ /// column silently took its own default value instead of being derived.
+ @Test(dataProvider = "bothV1AndV3")
+ public void
testDerivedColumnReadingDefaultColumnAddedInSameReload(SegmentVersion
segmentVersion)
+ throws Exception {
+ buildSegment(segmentVersion);
+
+ // newIntMetric is NOT derived -- it is a plain new column filled with its
default value (0).
+ _ingestionConfig.setTransformConfigs(
+ List.of(new TransformConfig(NEW_INT_SV_DIMENSION_COLUMN_NAME,
"plus(newIntMetric, 1)")));
+ runPreProcessor(_newColumnsSchema1);
+
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(INDEX_DIR);
+ // newIntMetric declares defaultNullValue 1 in the schema, so every row of
it is 1.
+ ColumnMetadata source =
segmentMetadata.getColumnMetadataFor(NEW_INT_METRIC_COLUMN_NAME);
+ assertEquals(source.getMinValue(), 1);
+
+ // Derived from that value (1 + 1 = 2). Had the argument not been visible,
the column would instead hold its own
+ // default null value, which for an INT dimension is Integer.MIN_VALUE.
+ ColumnMetadata derived =
segmentMetadata.getColumnMetadataFor(NEW_INT_SV_DIMENSION_COLUMN_NAME);
+ assertEquals(derived.getMinValue(), 2);
+ assertEquals(derived.getMaxValue(), 2);
+ assertTrue(derived.isAutoGenerated());
+ }
+
+ /// A column reading a cycle member must take the fallback even when that
member already exists in the segment and is
+ /// only being updated. Its metadata is then present, so nothing about a
missing argument would trigger the fallback,
+ /// and the new column -- ordered before the postponed cycle -- would read
the member's pre-update value for good.
+ @Test(dataProvider = "bothV1AndV3")
+ public void
testDerivedColumnReadingUpdatedCycleMemberFallsBackToDefaultValue(SegmentVersion
segmentVersion)
+ throws Exception {
+ buildSegment(segmentVersion);
+
+ // First reload: a two-column cycle, so both columns are created with
their schema default values (1 and 2).
+ _ingestionConfig.setTransformConfigs(
+ List.of(new TransformConfig(NEW_INT_METRIC_COLUMN_NAME,
"plus(newLongMetric, 1)"),
+ new TransformConfig(NEW_LONG_METRIC_COLUMN_NAME,
"plus(newIntMetric, 1)")));
+ Schema firstSchema = new Schema();
+ firstSchema.setSchemaName(_schema.getSchemaName());
+ firstSchema.addField(new MetricFieldSpec(NEW_INT_METRIC_COLUMN_NAME,
DataType.INT, 1));
+ firstSchema.addField(new MetricFieldSpec(NEW_LONG_METRIC_COLUMN_NAME,
DataType.LONG, 2L));
+ runPreProcessor(firstSchema);
+ SegmentMetadataImpl firstMetadata = new SegmentMetadataImpl(INDEX_DIR);
+
assertEquals(firstMetadata.getColumnMetadataFor(NEW_INT_METRIC_COLUMN_NAME).getMinValue(),
1);
+
assertEquals(firstMetadata.getColumnMetadataFor(NEW_LONG_METRIC_COLUMN_NAME).getMinValue(),
2L);
+
+ // Second reload: the cycle members' defaults change (an update, not an
add) and a column reading one is added.
+ _ingestionConfig.setTransformConfigs(
+ List.of(new TransformConfig(NEW_INT_METRIC_COLUMN_NAME,
"plus(newLongMetric, 1)"),
+ new TransformConfig(NEW_LONG_METRIC_COLUMN_NAME,
"plus(newIntMetric, 1)"),
+ new TransformConfig(NEW_DOUBLE_METRIC_COLUMN_NAME,
"plus(newIntMetric, 1)")));
+ Schema secondSchema = new Schema();
+ secondSchema.setSchemaName(_schema.getSchemaName());
+ secondSchema.addField(new MetricFieldSpec(NEW_INT_METRIC_COLUMN_NAME,
DataType.INT, 10));
+ secondSchema.addField(new MetricFieldSpec(NEW_LONG_METRIC_COLUMN_NAME,
DataType.LONG, 20L));
+ secondSchema.addField(new MetricFieldSpec(NEW_DOUBLE_METRIC_COLUMN_NAME,
DataType.DOUBLE, 0.0));
+ runPreProcessor(secondSchema);
+
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(INDEX_DIR);
+
assertEquals(segmentMetadata.getColumnMetadataFor(NEW_INT_METRIC_COLUMN_NAME).getMinValue(),
10);
+
assertEquals(segmentMetadata.getColumnMetadataFor(NEW_LONG_METRIC_COLUMN_NAME).getMinValue(),
20L);
+ // Its own default, not 1 + the cycle member's pre-update value of 1.
+ ColumnMetadata dependent =
segmentMetadata.getColumnMetadataFor(NEW_DOUBLE_METRIC_COLUMN_NAME);
+ assertEquals(dependent.getCardinality(), 1);
+ assertEquals(dependent.getMinValue(), 0.0);
+ }
+
+ /// A dependency cycle cannot be satisfied in any order. Both columns must
still be created, each falling back to
+ /// its default value, rather than failing the segment load.
+ @Test(dataProvider = "bothV1AndV3")
+ public void testDerivedColumnCycleFallsBackToDefaultValue(SegmentVersion
segmentVersion)
+ throws Exception {
+ buildSegment(segmentVersion);
+
+ _ingestionConfig.setTransformConfigs(
+ List.of(new TransformConfig(NEW_INT_METRIC_COLUMN_NAME,
"plus(newIntSVDimension, 1)"),
+ new TransformConfig(NEW_INT_SV_DIMENSION_COLUMN_NAME,
"plus(newIntMetric, 1)")));
+ runPreProcessor(_newColumnsSchema1);
+
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(INDEX_DIR);
+ for (String column : List.of(NEW_INT_METRIC_COLUMN_NAME,
NEW_INT_SV_DIMENSION_COLUMN_NAME)) {
+ ColumnMetadata columnMetadata =
segmentMetadata.getColumnMetadataFor(column);
+ assertNotNull(columnMetadata, column + " must still be created");
+ assertEquals(columnMetadata.getCardinality(), 1, column + " must hold a
single default value");
+ // Its own default value, not one derived from a sibling that was itself
defaulted: whichever member of the
+ // cycle is created second can see the first through the metadata
written a moment ago.
+ assertEquals(columnMetadata.getMinValue(),
_newColumnsSchema1.getFieldSpecFor(column).getDefaultNullValue(),
+ column + " must hold its own default value");
+ }
+ }
+
+ /// A derived column reading a source whose forward index is disabled. The
handler regenerates that forward index
+ /// from the dictionary and inverted index for the duration of the reload,
then drops it again.
+ ///
+ /// v3 only: a v1 reload cannot remove a forward index, so the source never
reaches the shape under test.
+ @Test
+ public void testDerivedColumnFromForwardIndexDisabledSource()
+ throws Exception {
+ buildSegment(SegmentVersion.v3);
+
+ // First reload: take the forward index away from the source column,
keeping its dictionary and inverted index.
+ _invertedIndexColumns.add(COLUMN1_NAME);
+ _fieldConfigMap.put(COLUMN1_NAME,
+ new FieldConfig(COLUMN1_NAME, FieldConfig.EncodingType.DICTIONARY,
List.of(), null,
+ Map.of(FieldConfig.FORWARD_INDEX_DISABLED, "true")));
+ runPreProcessor(_schema);
+ validateIndexDoesNotExist(StandardIndexes.forward(), COLUMN1_NAME);
+ validateIndexExists(StandardIndexes.dictionary(), COLUMN1_NAME);
+ validateIndexExists(StandardIndexes.inverted(), COLUMN1_NAME);
+
+ SegmentMetadataImpl beforeMetadata = new SegmentMetadataImpl(INDEX_DIR);
+ int sourceMin = (Integer)
beforeMetadata.getColumnMetadataFor(COLUMN1_NAME).getMinValue();
+ int sourceMax = (Integer)
beforeMetadata.getColumnMetadataFor(COLUMN1_NAME).getMaxValue();
+
+ // Second reload: derive from it.
+ _ingestionConfig.setTransformConfigs(
+ List.of(new TransformConfig(NEW_INT_SV_DIMENSION_COLUMN_NAME,
"plus(column1, 1)")));
+ runPreProcessor(_newColumnsSchema1);
+
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(INDEX_DIR);
+ ColumnMetadata derived =
segmentMetadata.getColumnMetadataFor(NEW_INT_SV_DIMENSION_COLUMN_NAME);
+ assertEquals(derived.getMinValue(), sourceMin + 1);
+ assertEquals(derived.getMaxValue(), sourceMax + 1);
+
+ // The regenerated forward index was temporary: the source keeps the shape
its config asks for.
+ validateIndexDoesNotExist(StandardIndexes.forward(), COLUMN1_NAME);
+ }
+
+ /// A derived column that itself has the forward index disabled: it is built
with one, and ForwardIndexHandler
+ /// drops it in its post-update cleanup once the inverted index has been
built from it.
+ ///
+ /// v3 only: a v1 reload cannot remove a forward index, so it stays on disk
there (same as every other
+ /// forward-index-disabled column on v1).
+ @Test
+ public void testForwardIndexDisabledDerivedColumn()
+ throws Exception {
+ buildSegment(SegmentVersion.v3);
+
+ _invertedIndexColumns.add(NEW_RAW_STRING_SV_DIMENSION_COLUMN_NAME);
+ _fieldConfigMap.put(NEW_RAW_STRING_SV_DIMENSION_COLUMN_NAME,
+ new FieldConfig(NEW_RAW_STRING_SV_DIMENSION_COLUMN_NAME,
FieldConfig.EncodingType.DICTIONARY, List.of(), null,
+ Map.of(FieldConfig.FORWARD_INDEX_DISABLED, "true")));
+ _ingestionConfig.setTransformConfigs(
+ List.of(new TransformConfig(NEW_RAW_STRING_SV_DIMENSION_COLUMN_NAME,
"reverse(column3)")));
+ runPreProcessor(_newColumnsSchema1);
+
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(INDEX_DIR);
+ ColumnMetadata derived =
segmentMetadata.getColumnMetadataFor(NEW_RAW_STRING_SV_DIMENSION_COLUMN_NAME);
+ assertTrue(derived.isAutoGenerated());
+ assertTrue(derived.hasDictionary(), "A forward-index-disabled column must
be dictionary encoded");
+ // Derived from column3, not filled with the default value: column3 has 5
distinct values.
+ assertEquals(derived.getCardinality(), 5);
+
+ validateIndexExists(StandardIndexes.dictionary(),
NEW_RAW_STRING_SV_DIMENSION_COLUMN_NAME);
+ validateIndexExists(StandardIndexes.inverted(),
NEW_RAW_STRING_SV_DIMENSION_COLUMN_NAME);
+ validateIndexDoesNotExist(StandardIndexes.forward(),
NEW_RAW_STRING_SV_DIMENSION_COLUMN_NAME);
+ }
+
+ /// A chain must give the same answer whether it is built at ingestion or at
load. Ingestion evaluates every
+ /// expression before anything is converted to its stored type, so an
intermediate link that loses precision on
+ /// the way to disk -- `INT newIntSVDimension = divide(column1, 2)` -- must
still hand the exact quotient to
+ /// `DOUBLE newDoubleMetric = plus(newIntSVDimension, 1)`. Reading the
intermediate back from its index would
+ /// quietly give the truncated value instead.
+ @Test(dataProvider = "bothV1AndV3")
+ public void testChainedDerivedColumnsMatchIngestion(SegmentVersion
segmentVersion)
+ throws Exception {
+ Schema schema = copyOfSchema(_schema);
+
schema.addField(_newColumnsSchema1.getFieldSpecFor(NEW_INT_SV_DIMENSION_COLUMN_NAME));
+
schema.addField(_newColumnsSchema1.getFieldSpecFor(NEW_DOUBLE_METRIC_COLUMN_NAME));
+ List<TransformConfig> transformConfigs =
+ List.of(new TransformConfig(NEW_DOUBLE_METRIC_COLUMN_NAME,
"plus(newIntSVDimension, 1)"),
+ new TransformConfig(NEW_INT_SV_DIMENSION_COLUMN_NAME,
"divide(column1, 2)"));
+
+ // The ingestion path: both columns are produced while the segment is
generated.
+ _ingestionConfig.setTransformConfigs(transformConfigs);
+ buildSegment(segmentVersion, schema);
+ List<Object> ingestedIntermediate =
readColumnValues(NEW_INT_SV_DIMENSION_COLUMN_NAME);
+ List<Object> ingestedDerived =
readColumnValues(NEW_DOUBLE_METRIC_COLUMN_NAME);
+
+ // The load path: the same two columns, built by the default column
handler on a segment that has neither.
+ _ingestionConfig.setTransformConfigs(null);
+ buildSegment(segmentVersion);
+ _ingestionConfig.setTransformConfigs(transformConfigs);
+ runPreProcessor(schema);
+
+ assertEquals(readColumnValues(NEW_INT_SV_DIMENSION_COLUMN_NAME),
ingestedIntermediate);
+ assertEquals(readColumnValues(NEW_DOUBLE_METRIC_COLUMN_NAME),
ingestedDerived);
+
+ // Guard against the comparison passing for the wrong reason: the two
paths can only differ on a row whose
+ // quotient has a fractional part, so the data has to contain at least one.
+ boolean anyTruncated = false;
+ for (int i = 0; i < ingestedDerived.size(); i++) {
+ if ((Double) ingestedDerived.get(i) != (Integer)
ingestedIntermediate.get(i) + 1.0) {
+ anyTruncated = true;
+ break;
+ }
+ }
+ assertTrue(anyTruncated, "Expected at least one row whose intermediate
value is truncated on the way to disk");
+ }
+
+ /// Regenerating a source's forward index can change the metadata its reader
has to work from. Here the encoding
+ /// flips from dictionary to raw in the very reload that derives from the
column, so the metadata captured before
+ /// the rebuild would open raw bytes with a dictionary reader.
+ ///
+ /// v3 only: a v1 reload cannot remove a forward index, so the source never
reaches the shape under test.
+ @Test
+ public void testDerivedColumnFromSourceRegeneratedWithNewEncoding()
+ throws Exception {
+ buildSegment(SegmentVersion.v3);
+
+ // First reload: take the forward index away from the source, keeping its
dictionary and inverted index.
+ _invertedIndexColumns.add(COLUMN1_NAME);
+ _fieldConfigMap.put(COLUMN1_NAME,
+ new FieldConfig(COLUMN1_NAME, FieldConfig.EncodingType.DICTIONARY,
List.of(), null,
+ Map.of(FieldConfig.FORWARD_INDEX_DISABLED, "true")));
+ runPreProcessor(_schema);
+ validateIndexDoesNotExist(StandardIndexes.forward(), COLUMN1_NAME);
+
+ SegmentMetadataImpl beforeMetadata = new SegmentMetadataImpl(INDEX_DIR);
+ ColumnMetadata sourceMetadata =
beforeMetadata.getColumnMetadataFor(COLUMN1_NAME);
+ assertTrue(sourceMetadata.hasDictionary());
+ int sourceMin = (Integer) sourceMetadata.getMinValue();
+ int sourceMax = (Integer) sourceMetadata.getMaxValue();
+
+ // Second reload: bring the forward index back as a raw one and derive
from the column at the same time.
+ _fieldConfigMap.remove(COLUMN1_NAME);
+ _invertedIndexColumns.remove(COLUMN1_NAME);
+ _noDictionaryColumns.add(COLUMN1_NAME);
+ _ingestionConfig.setTransformConfigs(
+ List.of(new TransformConfig(NEW_INT_SV_DIMENSION_COLUMN_NAME,
"plus(column1, 1)")));
+ runPreProcessor(_newColumnsSchema1);
+
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(INDEX_DIR);
+ ColumnMetadata derived =
segmentMetadata.getColumnMetadataFor(NEW_INT_SV_DIMENSION_COLUMN_NAME);
+ assertEquals(derived.getMinValue(), sourceMin + 1);
+ assertEquals(derived.getMaxValue(), sourceMax + 1);
+ assertEquals(derived.getCardinality(), sourceMetadata.getCardinality());
+ }
+
+ /// A source whose own action is an update, not an add. Its values are only
final once that update has run, so the
+ /// chain reading it has to be ordered after it -- otherwise the pre-update
value is written into the chain for
+ /// good, and no later reload puts it right.
+ @Test(dataProvider = "bothV1AndV3")
+ public void testDerivedColumnOrderedAfterUpdatedSource(SegmentVersion
segmentVersion)
+ throws Exception {
+ buildSegment(segmentVersion);
+
+ // First reload: the source is added, holding the default value its schema
declares.
+ Schema firstSchema = new Schema();
+ firstSchema.setSchemaName(_schema.getSchemaName());
+ firstSchema.addField(new MetricFieldSpec(NEW_INT_METRIC_COLUMN_NAME,
DataType.INT, 1));
+ runPreProcessor(firstSchema);
+ assertEquals(new
SegmentMetadataImpl(INDEX_DIR).getColumnMetadataFor(NEW_INT_METRIC_COLUMN_NAME).getMinValue(),
1);
+
+ // Second reload: the source's default changes and a two-link chain
reading it is added in the same run.
+ Schema secondSchema = new Schema();
+ secondSchema.setSchemaName(_schema.getSchemaName());
+ secondSchema.addField(new MetricFieldSpec(NEW_INT_METRIC_COLUMN_NAME,
DataType.INT, 2));
+ secondSchema.addField(new
DimensionFieldSpec(NEW_INT_SV_DIMENSION_COLUMN_NAME, DataType.INT, true));
+ secondSchema.addField(new MetricFieldSpec(NEW_LONG_METRIC_COLUMN_NAME,
DataType.LONG, 0L));
+ _ingestionConfig.setTransformConfigs(
+ List.of(new TransformConfig(NEW_LONG_METRIC_COLUMN_NAME,
"plus(newIntSVDimension, 1)"),
+ new TransformConfig(NEW_INT_SV_DIMENSION_COLUMN_NAME,
"plus(newIntMetric, 1)")));
+ runPreProcessor(secondSchema);
+
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(INDEX_DIR);
+
assertEquals(segmentMetadata.getColumnMetadataFor(NEW_INT_METRIC_COLUMN_NAME).getMinValue(),
2);
+
assertEquals(segmentMetadata.getColumnMetadataFor(NEW_INT_SV_DIMENSION_COLUMN_NAME).getMinValue(),
3);
+
assertEquals(segmentMetadata.getColumnMetadataFor(NEW_LONG_METRIC_COLUMN_NAME).getMinValue(),
4L);
+ }
+
+ /// A derived column reading a plain multi-value default column that is
added in the same reload with its forward
+ /// index disabled. That source has a dictionary and an inverted index but
no forward index, and it exists only in
+ /// the metadata properties written a moment ago -- the segment directory's
own metadata does not list it yet.
+ ///
+ /// v3 only: regenerating a forward index writes the index file and then
registers it under the same path on a v1
+ /// segment, so the rebuild cannot run there at all and the derived column
falls back to its default value.
+ @Test
+ public void
testDerivedColumnFromForwardIndexDisabledColumnAddedInSameReload()
+ throws Exception {
+ buildSegment(SegmentVersion.v3);
+
+ // A default value of its own, so a derived column that fell back to ITS
default would be visibly different.
+ Schema schema = copyOfSchema(_newColumnsSchema1);
+
schema.getFieldSpecFor(NEW_STRING_MV_DIMENSION_COLUMN_NAME).setDefaultNullValue("mvDefault");
+
+ _invertedIndexColumns.add(NEW_STRING_MV_DIMENSION_COLUMN_NAME);
+ _fieldConfigMap.put(NEW_STRING_MV_DIMENSION_COLUMN_NAME,
+ new FieldConfig(NEW_STRING_MV_DIMENSION_COLUMN_NAME,
FieldConfig.EncodingType.DICTIONARY, List.of(), null,
+ Map.of(FieldConfig.FORWARD_INDEX_DISABLED, "true")));
+ _ingestionConfig.setTransformConfigs(List.of(new
TransformConfig(NEW_RAW_STRING_SV_DIMENSION_COLUMN_NAME,
+ "arrayElementAtString(newStringMVDimension, 1)")));
+ runPreProcessor(schema);
+
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(INDEX_DIR);
+ ColumnMetadata derived =
segmentMetadata.getColumnMetadataFor(NEW_RAW_STRING_SV_DIMENSION_COLUMN_NAME);
+ assertTrue(derived.isAutoGenerated());
+ assertEquals(derived.getMinValue(), "mvDefault");
+ }
+
+ /// A chain whose intermediate link evaluates to null. Ingestion runs every
expression before any null is turned
+ /// into a column's default, so `newIntMetric = coalesce(newIntSVDimension,
7)` sees the null that
+ /// `newIntSVDimension = nullIf(column1, column1)` produced and stores 7,
while the intermediate itself lands on
+ /// disk as its own default (0). Handing the stored default to the rest of
the chain instead would store 0 in
+ /// `newIntMetric` for good -- a wrong value that no later reload corrects.
+ ///
+ /// Run with null handling on and off: the substitution happens either way,
the flag only decides whether a null
+ /// vector is written next to the intermediate.
+ @Test(dataProvider = "versionAndNullHandling")
+ public void
testChainedDerivedColumnsWithNullIntermediateMatchIngestion(SegmentVersion
segmentVersion,
+ boolean nullHandlingEnabled)
+ throws Exception {
+ _nullHandlingEnabled = nullHandlingEnabled;
+
+ Schema schema = copyOfSchema(_schema);
+ schema.addField(new DimensionFieldSpec(NEW_INT_SV_DIMENSION_COLUMN_NAME,
DataType.INT, true, 0));
+ schema.addField(new MetricFieldSpec(NEW_INT_METRIC_COLUMN_NAME,
DataType.INT, 0));
+ List<TransformConfig> transformConfigs =
+ List.of(new TransformConfig(NEW_INT_METRIC_COLUMN_NAME,
"coalesce(newIntSVDimension, 7)"),
+ new TransformConfig(NEW_INT_SV_DIMENSION_COLUMN_NAME,
"nullIf(column1, column1)"));
+
+ // The ingestion path: both columns are produced while the segment is
generated.
+ _ingestionConfig.setTransformConfigs(transformConfigs);
+ buildSegment(segmentVersion, schema);
+ List<Object> ingestedIntermediate =
readColumnValues(NEW_INT_SV_DIMENSION_COLUMN_NAME);
+ List<Object> ingestedDerived =
readColumnValues(NEW_INT_METRIC_COLUMN_NAME);
+ // Guard against the comparison below passing for the wrong reason: the
chain has to have run on a null, which
+ // shows up as the intermediate holding its default while the column
reading it does not.
+ assertEquals(ingestedIntermediate.get(0), 0);
+ assertEquals(ingestedDerived.get(0), 7);
+
+ // The load path: the same two columns, built by the default column
handler on a segment that has neither.
+ _ingestionConfig.setTransformConfigs(null);
+ buildSegment(segmentVersion);
+ _ingestionConfig.setTransformConfigs(transformConfigs);
+ runPreProcessor(schema);
+
+ assertEquals(readColumnValues(NEW_INT_SV_DIMENSION_COLUMN_NAME),
ingestedIntermediate);
+ assertEquals(readColumnValues(NEW_INT_METRIC_COLUMN_NAME),
ingestedDerived);
+ }
+
+ /// A transform that reads the same forward-index-disabled source twice.
Regenerating that source's forward index
+ /// happens on the first read and brings it back as a raw one, so every
later read has to work from the refreshed
+ /// metadata -- the copy taken before the rebuild describes a
dictionary-encoded column and would make the second
+ /// read open raw bytes with a dictionary reader.
+ ///
+ /// v3 only: a v1 reload cannot remove a forward index, so the source never
reaches the shape under test.
+ @Test
+ public void testDerivedColumnReadingRegeneratedSourceTwice()
+ throws Exception {
+ buildSegment(SegmentVersion.v3);
+
+ // First reload: take the forward index away from the source, keeping its
dictionary and inverted index.
+ _invertedIndexColumns.add(COLUMN1_NAME);
+ _fieldConfigMap.put(COLUMN1_NAME,
+ new FieldConfig(COLUMN1_NAME, FieldConfig.EncodingType.DICTIONARY,
List.of(), null,
+ Map.of(FieldConfig.FORWARD_INDEX_DISABLED, "true")));
+ runPreProcessor(_schema);
+ validateIndexDoesNotExist(StandardIndexes.forward(), COLUMN1_NAME);
+
+ SegmentMetadataImpl beforeMetadata = new SegmentMetadataImpl(INDEX_DIR);
+ ColumnMetadata sourceMetadata =
beforeMetadata.getColumnMetadataFor(COLUMN1_NAME);
+ assertTrue(sourceMetadata.hasDictionary());
+ double sourceMin = (Integer) sourceMetadata.getMinValue();
+ double sourceMax = (Integer) sourceMetadata.getMaxValue();
+
+ // Second reload: bring the forward index back as a raw one and read the
column twice in the same expression.
+ _fieldConfigMap.remove(COLUMN1_NAME);
+ _invertedIndexColumns.remove(COLUMN1_NAME);
+ _noDictionaryColumns.add(COLUMN1_NAME);
+ // A DOUBLE target, so that doubling the largest value of column1 stays
representable and a wrong read shows up
+ // as a wrong value rather than as both reads saturating at
Integer.MAX_VALUE.
+ _ingestionConfig.setTransformConfigs(
+ List.of(new TransformConfig(NEW_DOUBLE_METRIC_COLUMN_NAME,
"plus(column1, column1)")));
+ runPreProcessor(_newColumnsSchema1);
+
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(INDEX_DIR);
+ ColumnMetadata derived =
segmentMetadata.getColumnMetadataFor(NEW_DOUBLE_METRIC_COLUMN_NAME);
+ assertNotNull(derived, "The derived column must be created");
+ assertTrue(derived.isAutoGenerated());
+ assertEquals(derived.getMinValue(), sourceMin * 2.0);
+ assertEquals(derived.getMaxValue(), sourceMax * 2.0);
+ assertEquals(derived.getCardinality(), sourceMetadata.getCardinality());
+ }
+
+ /// Overlapping cycles: four columns that all reach each other, wired so
that a depth-first search closes its
+ /// first back edge over only part of them whichever column it starts from.
Reaching the rest of a cycle through
+ /// an edge into an already finished column is what a search like that
cannot see, and every one of these four
+ /// columns is reached that way from some starting point.
+ ///
+ /// Each has to hold its own default value. Deriving one of them from a
sibling that was itself defaulted would
+ /// store a value that looks real -- 12 rather than 22, say -- instead of
the documented fallback.
+ @Test(dataProvider = "bothV1AndV3")
+ public void
testOverlappingDerivedColumnCyclesFallBackToDefaultValue(SegmentVersion
segmentVersion)
+ throws Exception {
+ buildSegment(segmentVersion);
+
+ Schema schema = new Schema();
+ schema.setSchemaName(_schema.getSchemaName());
+ schema.addField(new DimensionFieldSpec(NEW_INT_SV_DIMENSION_COLUMN_NAME,
DataType.INT, true, 11));
+ schema.addField(new MetricFieldSpec(NEW_INT_METRIC_COLUMN_NAME,
DataType.INT, 22));
+ schema.addField(new MetricFieldSpec(NEW_LONG_METRIC_COLUMN_NAME,
DataType.LONG, 33L));
+ schema.addField(new MetricFieldSpec(NEW_DOUBLE_METRIC_COLUMN_NAME,
DataType.DOUBLE, 44.0));
+ _ingestionConfig.setTransformConfigs(List.of(
+ new TransformConfig(NEW_INT_SV_DIMENSION_COLUMN_NAME,
"plus(newIntMetric, newLongMetric)"),
+ new TransformConfig(NEW_INT_METRIC_COLUMN_NAME,
"plus(newIntSVDimension, newDoubleMetric)"),
+ new TransformConfig(NEW_LONG_METRIC_COLUMN_NAME, "plus(newIntMetric,
1)"),
+ new TransformConfig(NEW_DOUBLE_METRIC_COLUMN_NAME,
"plus(newIntSVDimension, 1)")));
+ runPreProcessor(schema);
+
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(INDEX_DIR);
+ for (String column : List.of(NEW_INT_SV_DIMENSION_COLUMN_NAME,
NEW_INT_METRIC_COLUMN_NAME,
+ NEW_LONG_METRIC_COLUMN_NAME, NEW_DOUBLE_METRIC_COLUMN_NAME)) {
+ ColumnMetadata columnMetadata =
segmentMetadata.getColumnMetadataFor(column);
+ assertNotNull(columnMetadata, column + " must still be created");
+ assertEquals(columnMetadata.getCardinality(), 1, column + " must hold a
single default value");
+ assertEquals(columnMetadata.getMinValue(),
schema.getFieldSpecFor(column).getDefaultNullValue(),
+ column + " must hold its own default value");
+ }
+ }
+
+ /// A deep copy, so a test can change a default value without the schema
fields -- read once in the constructor --
+ /// carrying that change into every test that runs after it.
+ private static Schema copyOfSchema(Schema schema)
+ throws IOException {
+ return Schema.fromString(schema.toSingleLineJsonString());
+ }
+
+ private static List<Object> readColumnValues(String column)
+ throws Exception {
+ List<Object> values = new ArrayList<>();
+ try (PinotSegmentRecordReader recordReader = new
PinotSegmentRecordReader(INDEX_DIR)) {
+ GenericRow reuse = new GenericRow();
+ while (recordReader.hasNext()) {
+ reuse.clear();
+ values.add(recordReader.next(reuse).getValue(column));
+ }
+ }
+ return values;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
