Jackie-Jiang commented on code in PR #19581:
URL: https://github.com/apache/pinot/pull/19581#discussion_r4289143406
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java:
##########
@@ -607,13 +692,20 @@ private void createDerivedColumnV1Indices(String column,
FunctionEvaluator funct
nullValueVectorCreator.seal();
}
+ // Take the untouched expression results aside before the switch below
converts them in place to the stored
+ // type, so the rest of the chain sees what ingestion would have given
it.
+ Object[] expressionValues = copyValuesForChain(column, outputValues);
Review Comment:
**[P1] Preserve null intermediate results before default substitution**
This copy is taken after `outputValue == null` has been replaced with the
column's default at lines 672–673, so it still does not preserve the expression
results ingestion passes to the rest of the chain. For example, with INT `b =
nullIf(a, a)` and default `0`, followed by INT `c = coalesce(b, 7)`, ingestion
evaluates `c` while `b` is null and stores `c=7`; this reload path caches `b=0`
and permanently stores `c=0`. Capture the chain values immediately after
expression evaluation and before null substitution. Please extend the
ingestion/reload comparison to null intermediates, with null handling both
enabled and disabled.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java:
##########
@@ -1219,6 +1315,299 @@ private void putDictionaryCompressionStats(String
column, long uncompressedValue
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;
+ }
+
+ Map<String, DefaultColumnAction> ordered = new
LinkedHashMap<>(actionMap.size());
+ Set<String> visiting = new LinkedHashSet<>();
+ for (String column : actionMap.keySet()) {
+ visitForOrdering(column, actionMap, dependencies, visiting, 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, Set<String> visiting,
Map<String, DefaultColumnAction> ordered) {
+ if (ordered.containsKey(column)) {
Review Comment:
**[P2] Identify all members of overlapping dependency cycles**
Returning for an already ordered node misses cycle members outside the first
DFS back-edge path. For `a = plus(b, c)`, `b = plus(a, 1)`, and `c = plus(b,
1)`, with all three defaults set to `0`, traversal from `a` marks `a,b` cyclic
and finishes `b`. When it visits `c`, the edge to `b` returns here, leaving `c`
unmarked even though `c -> b -> a -> c` is a cycle. The resulting order builds
`b=0`, then incorrectly derives `c=1`, then defaults `a`. Identify complete
strongly connected components before applying the fallback, and add an
overlapping-cycle regression asserting each member's own default value.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java:
##########
@@ -1219,6 +1315,299 @@ private void putDictionaryCompressionStats(String
column, long uncompressedValue
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;
+ }
+
+ Map<String, DefaultColumnAction> ordered = new
LinkedHashMap<>(actionMap.size());
+ Set<String> visiting = new LinkedHashSet<>();
+ for (String column : actionMap.keySet()) {
+ visitForOrdering(column, actionMap, dependencies, visiting, 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, Set<String> visiting,
Map<String, DefaultColumnAction> ordered) {
+ if (ordered.containsKey(column)) {
+ return;
+ }
+ if (!visiting.add(column)) {
+ // The cycle is the tail of the traversal from this column onwards.
Every one of its members has to be recorded:
+ // emitting them in any order leaves one reading a sibling that was
itself defaulted, which would silently
+ // derive a value from a default instead of taking the documented
fallback.
+ List<String> cycle = new ArrayList<>();
+ boolean inCycle = false;
+ for (String visited : visiting) {
+ inCycle |= visited.equals(column);
+ if (inCycle) {
+ cycle.add(visited);
+ }
+ }
+ 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.", cycle);
+ _cyclicColumns.addAll(cycle);
+ return;
+ }
+ try {
+ for (String dependency : dependencies.getOrDefault(column, List.of())) {
+ visitForOrdering(dependency, actionMap, dependencies, visiting,
ordered);
+ }
+ // A dependency stuck in a cycle is absent from `ordered`; emitting this
column anyway is safe because the
+ // missing argument makes it take the default-value path.
+ ordered.put(column, actionMap.get(column));
+ } finally {
+ visiting.remove(column);
+ }
+ }
+
+ /// 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);
+ }
+ }
+
+ /// A copy of the values a derived column produced, 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
+ /// the copy. The caller converts the original array in place, which is why
this is taken before that happens.
+ @Nullable
+ private Object[] copyValuesForChain(String column, Object[] outputValues) {
+ if (_transformConfigsByColumn == null) {
+ buildTransformConfigIndex();
+ }
+ return _derivedColumnArgumentNames.contains(column) ?
Arrays.copyOf(outputValues, outputValues.length) : 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);
+ _createdColumnMetadata.computeIfPresent(column, (k, v) -> refreshed);
Review Comment:
**[P1] Cache refreshed metadata for existing source columns too**
`computeIfPresent` does nothing for a source that already existed before
this reload, because it is absent from `_createdColumnMetadata`. Its first read
receives refreshed metadata, but the next `getColumnMetadata()` returns the
original `_segmentMetadata`, and `materializeSourceForwardIndex()` then returns
early because the temporary forward index already exists. Even one transform
such as `plus(source, source)` triggers this: the evaluator records both
identifier occurrences. If regeneration changes DICTIONARY encoding to RAW, the
second occurrence reads raw bytes with dictionary metadata; regenerated MV
sources with duplicates likewise retain incorrect entry counts on subsequent
reads. Cache every refreshed source with `put` and cover repeated references or
multiple consumers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
