Re: [PR] feat: restrict virtual column usage in clustered segments, allow remapping them at query time (druid)

["FrankChen021 (via GitHub)" <gi...@apache.org>]

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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org