FrankChen021 commented on code in PR #19659:
URL: https://github.com/apache/druid/pull/19659#discussion_r3637950221
##########
processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndexCursorFactory.java:
##########
@@ -154,19 +154,35 @@ private CursorHolder
makeClusteredCursorHolder(CursorBuildSpec spec, OnHeapClust
final List<Supplier<CursorHolder>> holderSuppliers = new
ArrayList<>(surviving.size());
final Closer closer = Closer.create();
for (TableClusterGroupSpec valueGroup : surviving) {
- final OnHeapClusterGroup group =
clusteredBaseTable.getGroupForClusteringValues(
- valueGroup.lookupClusteringValues()
- );
+ final Object[] groupClusteringValues =
valueGroup.lookupClusteringValues();
+ final OnHeapClusterGroup group =
clusteredBaseTable.getGroupForClusteringValues(groupClusteringValues);
if (group == null) {
throw DruidException.defensive(
"No cluster group for clustering values [%s]",
- Arrays.toString(valueGroup.lookupClusteringValues())
+ Arrays.toString(groupClusteringValues)
);
}
- clusteringValuesByGroup.add(valueGroup.lookupClusteringValues());
+ clusteringValuesByGroup.add(groupClusteringValues);
final CursorBuildSpec groupSpec = plan.rebuildCursorBuildSpec(spec,
valueGroup);
+ // Expose this group's clustering columns as constants to the per-group
cursor's selector factory
holderSuppliers.add(
- Suppliers.memoize(() -> closer.register(new
IncrementalIndexCursorHolder(group, groupSpec)))
+ Suppliers.memoize(() -> closer.register(
+ new IncrementalIndexCursorHolder(group, groupSpec)
+ {
+ @Override
+ public ColumnSelectorFactory makeSelectorFactory(
+ CursorBuildSpec buildSpec,
+ IncrementalIndexRowHolder currEntry
+ )
+ {
+ return new ClusteringColumnSelectorFactory(
Review Comment:
[P1] Expose clustering constants while evaluating realtime VCs
Wrapping `IncrementalIndexColumnSelectorFactory` here is too late for
virtual-column dependencies: that factory resolves a retained VC such as
`upper(tenant)` through itself, so its lookup of clustering column `tenant`
bypasses this outer wrapper and sees the per-group omitted column as null
instead of the clustering constant. An unrewritable filter on that VC can
therefore silently reject matching realtime rows. Build the VC-aware factory on
top of the clustering-aware base, or otherwise route VC dependency selectors
through the clustering wrapper.
##########
processing/src/main/java/org/apache/druid/segment/projections/Projections.java:
##########
@@ -681,13 +703,122 @@ public static ClusterGroupQueryPlan
planClusterGroupQuery(
kept.add(group);
}
}
- return new ClusterGroupQueryPlan(kept, rewriteCache::get);
+ return new ClusterGroupQueryPlan(kept, rewriteCache::get,
virtualColumnRemap);
+ }
+
+ /**
+ * Build a query-level remap of {@code queryVirtualColumnOutputName ->
materializedColumnName} for each query virtual
+ * column that has an equivalent materialized column in the clustered base
table (a clustering column produced by a
+ * group virtual column, or a non-clustering materialized virtual-column
output).
+ * <p>
+ * A remap target must be a column the per-group cursor can actually serve,
so {@code materializedColumns} restricts
+ * candidates to the summary's stored columns (clustering columns included
by construction). Group virtual columns
+ * whose output name is not a stored column are metadata-only carriers;
notably the {@code __virtualGranularity}
+ * query-granularity carrier, and are never valid substitution targets; a
query VC equivalent to such a carrier is
+ * left in place to recompute (e.g. from {@code __time}) rather than
remapped to an unreadable column.
+ * <p>
+ * The substitution is always a pure optimisation: {@code
ClusteredValueGroupsBaseTableProjectionSpec} requires every
+ * clustered virtual column's inputs to be stored columns, so a query VC
equivalent to one can always be recomputed
+ * from stored columns and reading the materialized column merely skips that
recomputation.
+ * <p>
+ * A query VC that {@code queryFilter} references is left unremapped when
the filter can't rewrite its required
+ * columns ({@link Filter#supportsRequiredColumnRewrite()} is false, e.g.
spatial / javascript / column-comparison
+ * filters): the filter can't have the VC name swapped for the materialized
column, and since a remappable VC is
+ * (per the spec) recomputable from stored columns, keeping it lets the
filter read the recomputed value instead of
+ * throwing on an unsupported rewrite.
+ * <p>
+ * A substituted (dropped) query virtual column is read from its
materialized column and never recomputes, so it
+ * imposes no requirement on its own inputs. A query virtual column must
therefore be kept (recomputed, not
+ * substituted) only when it is transitively required by a kept query
virtual column (because query VCs are computed
+ * in the per-group cursor, below the concat-level remap, so a dropped input
a kept VC still references would
+ * incorrectly resolve to null.)
+ */
+ private static Map<String, String> buildClusterVirtualColumnRemap(
+ VirtualColumns queryVcs,
+ VirtualColumns groupVcs,
+ Set<String> materializedColumns,
+ @Nullable Filter queryFilter
+ )
+ {
+ final VirtualColumn[] all = queryVcs.getVirtualColumns();
+ if (all.length == 0) {
+ return Map.of();
+ }
+ // A filter that can't rewrite its required columns (spatial, javascript,
column-comparison, ...) can't have a
+ // remapped VC name swapped for the materialized column in its own
required-column set, so any VC it references
+ // must stay in place. ClusteredValueGroupsBaseTableProjectionSpec
guarantees every clustered VC is recomputable
Review Comment:
[P1] Validate the VC invariant in persisted schemas
This recomputation path assumes every clustered VC input is stored, but only
`ClusteredValueGroupsBaseTableProjectionSpec` enforces that invariant.
`ClusteredValueGroupsBaseTableSchema` is deserialized directly from segment
metadata without the same check, so an older segment can omit a VC input and
this fallback will recompute from a missing column, silently misfiltering rows.
Validate or normalize the invariant when loading the schema, or avoid
recomputation unless the required inputs are present.
--
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]
