capistrant commented on code in PR #19659:
URL: https://github.com/apache/druid/pull/19659#discussion_r3538106431
##########
processing/src/test/java/org/apache/druid/segment/QueryableIndexCursorFactoryClusteredTest.java:
##########
@@ -182,6 +190,267 @@ private QueryableIndex standardTwoGroup()
));
}
+ /**
+ * Cluster spec for a segment clustered on {@code tenant_lower :=
lower(tenant)} (a clustering column produced by a
+ * group VC; raw {@code tenant} is NOT a stored column) plus a
non-clustering materialized
+ * {@code region_upper := upper(region)} column. Columns: {@code
[tenant_lower (clustering), region, region_upper,
+ * __time]}.
+ */
+ private static final ClusteredValueGroupsBaseTableProjectionSpec
VIRTUAL_CLUSTER_SPEC =
+ ClusteredValueGroupsBaseTableProjectionSpec.builder()
+ .virtualColumns(VirtualColumns.create(
+ new ExpressionVirtualColumn("tenant_lower", "lower(tenant)",
ColumnType.STRING, TestExprMacroTable.INSTANCE),
+ new ExpressionVirtualColumn("region_upper", "upper(region)",
ColumnType.STRING, TestExprMacroTable.INSTANCE)
+ ))
+ .columns(
+ new StringDimensionSchema("tenant_lower"),
+ StringDimensionSchema.create("region"),
+ StringDimensionSchema.create("region_upper"),
+ new LongDimensionSchema("__time")
+ )
+ .clusteringColumns("tenant_lower")
+ .build();
+
+ private QueryableIndex buildVirtualClusteringSegment()
+ {
+ final IncrementalIndexSchema schema =
+ IncrementalIndexSchema.builder()
+ .withMinTimestamp(INTERVAL.getStartMillis())
+ .withTimestampSpec(new TimestampSpec("__time",
"auto", null))
+ .withQueryGranularity(Granularities.NONE)
+
.withDimensionsSpec(VIRTUAL_CLUSTER_SPEC.getDimensionsSpec())
+ .withRollup(false)
+ .withClusterSpec(VIRTUAL_CLUSTER_SPEC)
+ .build();
+ return IndexBuilder.create()
+ .useV10()
+ .tmpDir(tmpDir)
+ .schema(schema)
+ .rows(List.of(
+ row("Acme", "2025-01-01T00:00:00", "us-east-1"),
+ row("Acme", "2025-01-01T01:00:00", "us-west-2"),
+ row("Globex", "2025-01-01T00:30:00", "eu-west-1")
+ ))
+ .buildMMappedIndex(INTERVAL);
+ }
+
+ @Test
+ void
testQueryVcEquivalentToClusteringColumnReadsMaterializedColumnViaAsCursor()
+ {
+ // Raw `tenant` is NOT stored; query VC v0 := lower(tenant) is equivalent
to the clustering column tenant_lower.
+ // The scalar selector remap must substitute the materialized tenant_lower
clustering constant — recompute would
+ // be null. Read via asCursor() (non-vector) to exercise the scalar
ClusteringColumnSelectorFactory path.
+ segmentIndex = buildVirtualClusteringSegment();
+ final QueryableIndexCursorFactory factory = new
QueryableIndexCursorFactory(
+ segmentIndex,
+ QueryableIndexTimeBoundaryInspector.create(segmentIndex)
+ );
+ final CursorBuildSpec buildSpec = CursorBuildSpec.builder()
+ .setVirtualColumns(VirtualColumns.create(
+ new ExpressionVirtualColumn("v0", "lower(tenant)",
ColumnType.STRING, TestExprMacroTable.INSTANCE)
+ ))
+ .build();
+ try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) {
+ Assertions.assertEquals(List.of("acme", "acme", "globex"),
collectDimension(holder.asCursor(), "v0"));
+ }
+ }
+
+ @Test
+ void
testQueryVcEquivalentToNonClusteringMaterializedColumnReadsMaterializedColumnViaAsCursor()
+ {
+ // Query VC v1 := upper(region) is equivalent to the non-clustering
materialized column region_upper. The remap
+ // makes makeDimensionSelector("v1") read the per-group physical
region_upper column.
+ segmentIndex = buildVirtualClusteringSegment();
+ final QueryableIndexCursorFactory factory = new
QueryableIndexCursorFactory(
+ segmentIndex,
+ QueryableIndexTimeBoundaryInspector.create(segmentIndex)
+ );
+ final CursorBuildSpec buildSpec = CursorBuildSpec.builder()
+ .setVirtualColumns(VirtualColumns.create(
+ new ExpressionVirtualColumn("v1", "upper(region)",
ColumnType.STRING, TestExprMacroTable.INSTANCE)
+ ))
+ .build();
+ try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) {
+ Assertions.assertEquals(
+ List.of("US-EAST-1", "US-WEST-2", "EU-WEST-1"),
+ collectDimension(holder.asCursor(), "v1")
+ );
+ }
+ }
+
+ @Test
+ void testQueryVcWithNoEquivalentStillRecomputesViaAsCursor()
+ {
+ // No-regression: query VC with no materialized equivalent
(lower(region_upper)) is recomputed from the stored
+ // region_upper column, not remapped.
+ segmentIndex = buildVirtualClusteringSegment();
+ final QueryableIndexCursorFactory factory = new
QueryableIndexCursorFactory(
+ segmentIndex,
+ QueryableIndexTimeBoundaryInspector.create(segmentIndex)
+ );
+ final CursorBuildSpec buildSpec = CursorBuildSpec.builder()
+ .setVirtualColumns(VirtualColumns.create(
+ new ExpressionVirtualColumn("v2", "lower(region_upper)",
ColumnType.STRING, TestExprMacroTable.INSTANCE)
+ ))
+ .build();
+ try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) {
+ Assertions.assertEquals(
+ List.of("us-east-1", "us-west-2", "eu-west-1"),
+ collectDimension(holder.asCursor(), "v2")
+ );
+ }
+ }
+
+ @Test
+ void
testQueryVcEquivalentToClusteringColumnSingleGroupReadsMaterializedColumn()
+ {
+ // Single surviving group (filter on the equivalent VC prunes to one
group), exercising the single-group cursor
+ // holder path's scalar remap wrap. Raw `tenant` is not stored, so the
materialized clustering constant is the
+ // only correct source.
+ segmentIndex = buildVirtualClusteringSegment();
+ final QueryableIndexCursorFactory factory = new
QueryableIndexCursorFactory(
+ segmentIndex,
+ QueryableIndexTimeBoundaryInspector.create(segmentIndex)
+ );
+ final CursorBuildSpec buildSpec = CursorBuildSpec.builder()
+ .setVirtualColumns(VirtualColumns.create(
+ new ExpressionVirtualColumn("v0", "lower(tenant)",
ColumnType.STRING, TestExprMacroTable.INSTANCE)
+ ))
+ .setFilter(new EqualityFilter("v0", ColumnType.STRING, "globex", null))
+ .build();
+ try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) {
+ Assertions.assertEquals(List.of("globex"),
collectDimension(holder.asCursor(), "v0"));
+ }
+ }
+
+ @Test
+ void testEquivalentVcFilterPrunesToSingleGroupNotConcatenated()
+ {
+ // A filter on the ORIGINAL expression used to build the clustering column
(lower(tenant) = 'acme', planned as a
+ // query VC) must prune to the single matching cluster group and take the
single-group cursor path -- NOT survive
+ // every group and post-filter through the ConcatenatingCursor.
+ segmentIndex = buildVirtualClusteringSegment();
+ final QueryableIndexCursorFactory factory = new
QueryableIndexCursorFactory(
+ segmentIndex,
+ QueryableIndexTimeBoundaryInspector.create(segmentIndex)
+ );
+ final VirtualColumns queryVcs = VirtualColumns.create(
+ new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING,
TestExprMacroTable.INSTANCE)
+ );
+
+ // Control: unfiltered -> both groups survive -> multi-group
ConcatenatingCursor.
+ try (CursorHolder holder =
+
factory.makeCursorHolder(CursorBuildSpec.builder().setVirtualColumns(queryVcs).build()))
{
+ Assertions.assertInstanceOf(ConcatenatingCursor.class,
holder.asCursor());
+ }
+
+ // Filtered on the equivalent VC -> the clustering leaf resolves to
tenant_lower and folds, pruning to the single
+ // acme group -> single-group cursor (not a ConcatenatingCursor), and no
residual filter on the missing raw column.
+ final CursorBuildSpec buildSpec = CursorBuildSpec.builder()
+ .setVirtualColumns(queryVcs)
+ .setFilter(new EqualityFilter("v0", ColumnType.STRING, "acme", null))
+ .build();
+ try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) {
+ final Cursor cursor = holder.asCursor();
+ Assertions.assertFalse(cursor instanceof ConcatenatingCursor);
+ Assertions.assertEquals(List.of("us-east-1", "us-west-2"),
collectDimension(cursor, "region"));
+ }
+ }
+
+ @Test
+ void
testQueryVcEquivalentToClusteringColumnReadsMaterializedColumnViaVectorCursor()
+ {
+ // the query-VC -> materialized-column remap is applied on the vector
factory too, so a
+ // vectorized read of v0 := lower(tenant) resolves to the materialized
clustering column tenant_lower (raw tenant
+ // not stored -> recompute would be null), and the remap no longer forces
the scalar path.
+ segmentIndex = buildVirtualClusteringSegment();
+ final QueryableIndexCursorFactory factory = new
QueryableIndexCursorFactory(
+ segmentIndex,
+ QueryableIndexTimeBoundaryInspector.create(segmentIndex)
+ );
+ final CursorBuildSpec buildSpec = CursorBuildSpec.builder()
+ .setVirtualColumns(VirtualColumns.create(
+ new ExpressionVirtualColumn("v0", "lower(tenant)",
ColumnType.STRING, TestExprMacroTable.INSTANCE)
+ ))
+ .build();
+ try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) {
+ Assertions.assertTrue(holder.canVectorize());
+ Assertions.assertEquals(List.of("acme", "acme", "globex"),
collectObjectVector(holder.asVectorCursor(), "v0"));
+ }
+ }
+
+ @Test
+ void
testQueryVcEquivalentToNonClusteringMaterializedColumnReadsMaterializedColumnViaVectorCursor()
+ {
+ // non-clustering: v1 := upper(region) resolves to the materialized
region_upper physical column.
+ segmentIndex = buildVirtualClusteringSegment();
+ final QueryableIndexCursorFactory factory = new
QueryableIndexCursorFactory(
+ segmentIndex,
+ QueryableIndexTimeBoundaryInspector.create(segmentIndex)
+ );
+ final CursorBuildSpec buildSpec = CursorBuildSpec.builder()
+ .setVirtualColumns(VirtualColumns.create(
+ new ExpressionVirtualColumn("v1", "upper(region)",
ColumnType.STRING, TestExprMacroTable.INSTANCE)
+ ))
+ .build();
+ try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) {
+ Assertions.assertTrue(holder.canVectorize());
+ Assertions.assertEquals(
+ List.of("US-EAST-1", "US-WEST-2", "EU-WEST-1"),
+ collectObjectVector(holder.asVectorCursor(), "v1")
+ );
+ }
+ }
+
+ @Test
+ void testQueryVcEquivalentToClusteringColumnSingleGroupVectorCursor()
+ {
+ // single surviving group (filter on the equivalent VC prunes to one
group): exercises the single-group holder's
+ // vector remap wrap.
+ segmentIndex = buildVirtualClusteringSegment();
+ final QueryableIndexCursorFactory factory = new
QueryableIndexCursorFactory(
+ segmentIndex,
+ QueryableIndexTimeBoundaryInspector.create(segmentIndex)
+ );
+ final CursorBuildSpec buildSpec = CursorBuildSpec.builder()
+ .setVirtualColumns(VirtualColumns.create(
+ new ExpressionVirtualColumn("v0", "lower(tenant)",
ColumnType.STRING, TestExprMacroTable.INSTANCE)
+ ))
+ .setFilter(new EqualityFilter("v0", ColumnType.STRING, "globex", null))
+ .build();
+ try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) {
+ Assertions.assertTrue(holder.canVectorize());
+ Assertions.assertEquals(List.of("globex"),
collectObjectVector(holder.asVectorCursor(), "v0"));
+ }
+ }
+
+ private static List<String> collectDimension(Cursor cursor, String column)
+ {
+ final DimensionSelector sel =
+
cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of(column));
+ final List<String> out = new ArrayList<>();
+ while (!cursor.isDone()) {
+ out.add(sel.getRow().size() == 0 ? null :
sel.lookupName(sel.getRow().get(0)));
+ cursor.advance();
+ }
+ return out;
+ }
+
+ private static List<String> collectObjectVector(VectorCursor cursor, String
column)
+ {
+ final VectorObjectSelector sel =
cursor.getColumnSelectorFactory().makeObjectSelector(column);
+ final List<String> out = new ArrayList<>();
+ while (!cursor.isDone()) {
+ final Object[] vector = sel.getObjectVector();
+ final int size = cursor.getCurrentVectorSize();
+ for (int i = 0; i < size; i++) {
+ out.add((String) vector[i]);
+ }
+ cursor.advance();
+ }
+ return out;
+ }
Review Comment:
can we move to top or bottom of file to avoid interleaving with tests
##########
processing/src/main/java/org/apache/druid/segment/projections/ClusterGroupQueryPlan.java:
##########
@@ -76,21 +93,51 @@ public Filter rewriteFor(TableClusterGroupSpec group)
/**
* Rebuild {@code spec} for {@code group}'s per-group cursor by swapping in
this plan's per-group filter rewrite
- * (see {@link #rewriteFor}). Returns {@code spec} unchanged when there is
no filter, or when the rewrite is
- * identical to the original (no clustering leaves folded), so the common
no-op case allocates nothing. Shared by
- * the {@link org.apache.druid.segment.QueryableIndexCursorFactory}
(historical) and
- * {@link
org.apache.druid.segment.incremental.IncrementalIndexCursorFactory} (realtime)
clustered dispatch.
+ * (see {@link #rewriteFor}) and, when {@link #virtualColumnRemap()} is
non-empty, substituting any query virtual
+ * columns that are equivalent to a materialized column.
+ * <p>
+ * When there is a remap, the matched query virtual columns (the remap keys)
are dropped from the spec's virtual
+ * columns, and the per-group filter's required columns are rewritten via
the remap so that non-clustering-VC filter
+ * leaves point at the materialized physical column (clustering-VC leaves
were already folded to TRUE / FALSE by the
+ * per-group rewrite walk, so they won't reference the dropped virtual
columns). The grouping / select / aggregator
+ * references are served instead by the {@link
org.apache.druid.segment.RemapColumnSelectorFactory} the cursor
+ * factory wraps around the {@link ClusteringColumnSelectorFactory}.
Review Comment:
> clustering-VC leaves were already folded to TRUE / FALSE by the per-group
rewrite walk, so they won't reference the dropped virtual columns
This is a not entirely true. `walkClusterGroupFilter` only folds Equality,
In, and Null. Everything else falls through unchanged. This will result in
wrong results when something like a Range filter on a query vc with equivalent
clustering column survives the walk.
Repro wrong empty results:
* tenant_lower := lower(tenant) clustered vc
* v0 := lower(tenant) ⇒ remap v0 → tenant_lower
* same test tenants as always
* RangeFilter("v0", ["a","z"])
* Both groups survive and you expect all rows but get empty result back
##########
processing/src/main/java/org/apache/druid/segment/projections/ClusterGroupQueryPlan.java:
##########
@@ -76,21 +93,51 @@ public Filter rewriteFor(TableClusterGroupSpec group)
/**
* Rebuild {@code spec} for {@code group}'s per-group cursor by swapping in
this plan's per-group filter rewrite
- * (see {@link #rewriteFor}). Returns {@code spec} unchanged when there is
no filter, or when the rewrite is
- * identical to the original (no clustering leaves folded), so the common
no-op case allocates nothing. Shared by
- * the {@link org.apache.druid.segment.QueryableIndexCursorFactory}
(historical) and
- * {@link
org.apache.druid.segment.incremental.IncrementalIndexCursorFactory} (realtime)
clustered dispatch.
+ * (see {@link #rewriteFor}) and, when {@link #virtualColumnRemap()} is
non-empty, substituting any query virtual
+ * columns that are equivalent to a materialized column.
+ * <p>
+ * When there is a remap, the matched query virtual columns (the remap keys)
are dropped from the spec's virtual
+ * columns, and the per-group filter's required columns are rewritten via
the remap so that non-clustering-VC filter
+ * leaves point at the materialized physical column (clustering-VC leaves
were already folded to TRUE / FALSE by the
+ * per-group rewrite walk, so they won't reference the dropped virtual
columns). The grouping / select / aggregator
+ * references are served instead by the {@link
org.apache.druid.segment.RemapColumnSelectorFactory} the cursor
+ * factory wraps around the {@link ClusteringColumnSelectorFactory}.
*/
public CursorBuildSpec rebuildCursorBuildSpec(CursorBuildSpec spec,
TableClusterGroupSpec group)
{
- if (spec.getFilter() == null) {
- return spec;
+ if (virtualColumnRemap.isEmpty()) {
+ if (spec.getFilter() == null) {
+ return spec;
+ }
+ final Filter rewritten = rewriteFor(group);
+ if (rewritten == spec.getFilter()) {
+ return spec;
+ }
+ return CursorBuildSpec.builder(spec).setFilter(rewritten).build();
}
- final Filter rewritten = rewriteFor(group);
- if (rewritten == spec.getFilter()) {
- return spec;
+
+ // Drop the remapped query virtual columns from the per-group spec; the
materialized columns they map to are
+ // either the per-group constant clustering column or a per-group physical
column, both served directly by the
+ // ClusteringColumnSelectorFactory (the cursor factory additionally wraps
it with a RemapColumnSelectorFactory so
+ // the original query VC names resolve to the materialized columns).
+ final List<VirtualColumn> prunedVcs = new ArrayList<>();
Review Comment:
naming seems backwards. aren't these actually whare are not being pruned out
of the query?
--
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]
