Jackie-Jiang commented on code in PR #19371:
URL: https://github.com/apache/pinot/pull/19371#discussion_r3994481823
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MaterializedViewClusterIntegrationTest.java:
##########
@@ -926,6 +948,110 @@ public void
testScalarGroupingFunctionResidualOnSourceColumnFallsBack()
"Fallback should return the same complete, ordered result as the
rewrite-disabled query");
}
+ /// -----------------------------------------------------------------------
+ /// Sketch re-aggregation: CPC, theta, integer-sum tuple
+ /// -----------------------------------------------------------------------
+
+ /// The MV stores a raw CPC sketch of Origin per Carrier (populated from the
source's own raw
+ /// aggregation, so the bytes are identical). Verifies both the cardinality
result rule and the
+ /// raw-self rule are rewritten to the MV and return the same values as the
base table.
+ @Test
+ public void testCpcSketchRewrite()
+ throws Exception {
+ assertMaterializedViewMatchesBaseline(
+ "SELECT Carrier, DISTINCTCOUNTCPCSKETCH(Origin) FROM " +
SOURCE_TABLE_NAME + " GROUP BY Carrier",
+ MATERIALIZED_VIEW_SKETCH_TABLE_OFFLINE);
+ assertRawSketchMatchesBaseline(
+ "SELECT Carrier, DISTINCTCOUNTRAWCPCSKETCH(Origin) FROM " +
SOURCE_TABLE_NAME + " GROUP BY Carrier",
+ "SELECT Carrier,
GETCPCSKETCHESTIMATE(DISTINCTCOUNTRAWCPCSKETCH(Origin)) FROM " +
SOURCE_TABLE_NAME
+ + " GROUP BY Carrier", MATERIALIZED_VIEW_SKETCH_TABLE_OFFLINE);
+ }
+
+ @Test
+ public void testThetaSketchRewrite()
+ throws Exception {
+ assertMaterializedViewMatchesBaseline(
+ "SELECT Carrier, DISTINCTCOUNTTHETASKETCH(Origin) FROM " +
SOURCE_TABLE_NAME + " GROUP BY Carrier",
+ MATERIALIZED_VIEW_SKETCH_TABLE_OFFLINE);
+ assertRawSketchMatchesBaseline(
+ "SELECT Carrier, DISTINCTCOUNTRAWTHETASKETCH(Origin) FROM " +
SOURCE_TABLE_NAME + " GROUP BY Carrier",
+ "SELECT Carrier,
GETTHETASKETCHESTIMATE(DISTINCTCOUNTRAWTHETASKETCH(Origin)) FROM " +
SOURCE_TABLE_NAME
+ + " GROUP BY Carrier", MATERIALIZED_VIEW_SKETCH_TABLE_OFFLINE);
+ }
+
+ /// One stored raw integer-sum tuple sketch per group serves cardinality,
sum, average and the raw
+ /// result, since they share an accumulator and differ only in how the final
value is extracted.
+ @Test
+ public void testTupleSketchRewrite()
+ throws Exception {
+ assertMaterializedViewMatchesBaseline(
+ "SELECT tupleGroup, DISTINCTCOUNTTUPLESKETCH(tupleSketch) FROM " +
TUPLE_SOURCE_TABLE_NAME
+ + " GROUP BY tupleGroup", MATERIALIZED_VIEW_TUPLE_TABLE_OFFLINE);
+ assertMaterializedViewMatchesBaseline(
+ "SELECT tupleGroup, SUMVALUESINTEGERSUMTUPLESKETCH(tupleSketch) FROM "
+ TUPLE_SOURCE_TABLE_NAME
+ + " GROUP BY tupleGroup", MATERIALIZED_VIEW_TUPLE_TABLE_OFFLINE);
+ assertMaterializedViewMatchesBaseline(
+ "SELECT tupleGroup, AVGVALUEINTEGERSUMTUPLESKETCH(tupleSketch) FROM "
+ TUPLE_SOURCE_TABLE_NAME
+ + " GROUP BY tupleGroup", MATERIALIZED_VIEW_TUPLE_TABLE_OFFLINE);
+ assertRawSketchMatchesBaseline(
+ "SELECT tupleGroup, DISTINCTCOUNTRAWINTEGERSUMTUPLESKETCH(tupleSketch)
FROM " + TUPLE_SOURCE_TABLE_NAME
+ + " GROUP BY tupleGroup",
+ "SELECT tupleGroup,
GETINTTUPLESKETCHESTIMATE(DISTINCTCOUNTRAWINTEGERSUMTUPLESKETCH(tupleSketch))
FROM "
+ + TUPLE_SOURCE_TABLE_NAME + " GROUP BY tupleGroup",
MATERIALIZED_VIEW_TUPLE_TABLE_OFFLINE);
+ }
+
+ /// Runs a query with MV rewrite enabled (broker default) and again with it
disabled, asserting the
+ /// enabled run is served by the expected MV, the disabled run hits no MV,
and the per-group
+ /// results are identical (compared unordered, since neither query imposes
an ordering).
+ private void assertMaterializedViewMatchesBaseline(String query, String
expectedMaterializedViewOfflineTable)
+ throws Exception {
+ JsonNode withRewrite = postQuery(query);
+ assertNoExceptions(withRewrite);
+ assertEquals(getMaterializedViewQueried(withRewrite),
expectedMaterializedViewOfflineTable,
+ "Expected MV " + expectedMaterializedViewOfflineTable + " to be
selected. Response: " + withRewrite);
+
+ JsonNode baseline = postQuery("SET enableMaterializedViewRewrite=false;\n"
+ query);
+ assertNoExceptions(baseline);
+ assertNull(getMaterializedViewQueried(baseline), "Baseline query must not
hit any MV");
+
+ Map<String, JsonNode> rewriteValues = valuesByGroup(withRewrite);
+ Map<String, JsonNode> baselineValues = valuesByGroup(baseline);
+ assertFalse(rewriteValues.isEmpty(), "Rewrite result should have rows");
+ assertEquals(rewriteValues, baselineValues, "MV rewrite result must equal
the base-table baseline");
+ }
+
+ /// Maps the first result column (the group key) to the second (the
aggregation value), so two
+ /// result sets can be compared without depending on row order.
+ private static Map<String, JsonNode> valuesByGroup(JsonNode response) {
+ Map<String, JsonNode> valuesByGroup = new HashMap<>();
+ for (JsonNode row : response.get("resultTable").get("rows")) {
+ valuesByGroup.put(row.get(0).asText(), row.get(1));
+ }
+ return valuesByGroup;
+ }
+
+ /// Covers the raw-sketch shape (user wants the merged sketch itself). The
bare raw query asserts
+ /// the raw-self rule selects the MV; the estimate query wraps the same
aggregation in the family's
+ /// estimate scalar and checks the value equals the base-table baseline.
Estimates are used rather
+ /// than the raw bytes because merging one stored sketch re-serializes to
different (content-equal)
+ /// bytes than the base table's direct sketch.
+ private void assertRawSketchMatchesBaseline(String rawQuery, String
estimateQuery,
+ String expectedMaterializedViewOfflineTable)
+ throws Exception {
+ JsonNode rawResponse = postQuery(rawQuery);
+ assertNoExceptions(rawResponse);
+ assertEquals(getMaterializedViewQueried(rawResponse),
expectedMaterializedViewOfflineTable,
+ "Expected MV " + expectedMaterializedViewOfflineTable + " to be
selected. Response: " + rawResponse);
+ assertFalse(valuesByGroup(rawResponse).isEmpty(), "Raw sketch result
should have rows");
+
+ JsonNode withRewrite = postQuery(estimateQuery);
+ assertNoExceptions(withRewrite);
+ JsonNode baseline = postQuery("SET enableMaterializedViewRewrite=false;\n"
+ estimateQuery);
+ assertNoExceptions(baseline);
+ assertEquals(valuesByGroup(withRewrite), valuesByGroup(baseline),
+ "Raw sketch estimate must match the base-table baseline");
Review Comment:
**[P2] Compare the actual rewritten raw results.** The `GET*SKETCHESTIMATE`
wrappers do not match an aggregation equivalence, so both estimate queries fall
back to the source table. Their equality never validates the rewritten raw
response, which is only checked for MV selection and nonempty rows. Decode and
compare the actual raw responses with rewrite enabled and disabled, including
response types and tuple summaries.
##########
pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/rewrite/equivalence/AggregationEquivalenceRegistry.java:
##########
@@ -48,7 +48,23 @@ public final class AggregationEquivalenceRegistry {
new SketchMergeEquivalence("DISTINCTCOUNTHLL", "DISTINCTCOUNTRAWHLL",
"DISTINCTCOUNTHLL"),
new SketchMergeEquivalence("DISTINCTCOUNTHLLPLUS",
"DISTINCTCOUNTRAWHLLPLUS", "DISTINCTCOUNTHLLPLUS"),
new SketchMergeEquivalence("DISTINCTCOUNTTHETASKETCH",
"DISTINCTCOUNTRAWTHETASKETCH",
- "DISTINCTCOUNTTHETASKETCH")
+ "DISTINCTCOUNTTHETASKETCH"),
+ new SketchMergeEquivalence("DISTINCTCOUNTCPCSKETCH",
"DISTINCTCOUNTRAWCPCSKETCH",
+ "DISTINCTCOUNTCPCSKETCH"),
+ new SketchMergeEquivalence("DISTINCTCOUNTTUPLESKETCH",
"DISTINCTCOUNTRAWINTEGERSUMTUPLESKETCH",
+ "DISTINCTCOUNTTUPLESKETCH"),
+ new SketchMergeEquivalence("SUMVALUESINTEGERSUMTUPLESKETCH",
"DISTINCTCOUNTRAWINTEGERSUMTUPLESKETCH",
+ "SUMVALUESINTEGERSUMTUPLESKETCH"),
+ new SketchMergeEquivalence("AVGVALUEINTEGERSUMTUPLESKETCH",
"DISTINCTCOUNTRAWINTEGERSUMTUPLESKETCH",
+ "AVGVALUEINTEGERSUMTUPLESKETCH"),
+
+ /// Sketch-based: user wants the merged sketch itself, MV stores the
same raw sketch
+ new SketchMergeEquivalence("DISTINCTCOUNTRAWTHETASKETCH",
"DISTINCTCOUNTRAWTHETASKETCH",
+ "DISTINCTCOUNTRAWTHETASKETCH"),
+ new SketchMergeEquivalence("DISTINCTCOUNTRAWCPCSKETCH",
"DISTINCTCOUNTRAWCPCSKETCH",
+ "DISTINCTCOUNTRAWCPCSKETCH"),
+ new SketchMergeEquivalence("DISTINCTCOUNTRAWINTEGERSUMTUPLESKETCH",
"DISTINCTCOUNTRAWINTEGERSUMTUPLESKETCH",
+ "DISTINCTCOUNTRAWINTEGERSUMTUPLESKETCH")
Review Comment:
**[P1] Preserve raw-sketch result types on exact matches.** For a
full-coverage MV with the same projection and grouping,
`ExactSubsumptionStrategy` wins before these raw-self rules and projects the
stored `BYTES` column directly. CPC and tuple raw functions return base64
`STRING` results, so enabling rewrite changes the response type and encoding to
hexadecimal `BYTES`. Reject these exact plans and let raw reaggregation
preserve the function's output contract. Add raw-result schema and content
assertions.
##########
pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/rewrite/equivalence/SketchMergeEquivalence.java:
##########
@@ -90,4 +102,116 @@ public Expression rewrite(Expression userAggExpression,
String materializedViewC
return RequestUtils.getFunctionExpression(_reAggFunctionName.toLowerCase(),
operands.toArray(new Expression[0]));
}
+
+ @Override
+ public boolean operandsCompatible(@Nullable List<Expression> userOperands,
+ @Nullable List<Expression> materializedViewOperands) {
+ Family family = family();
+ if (family == null) {
+ return true;
+ }
+ if (family == Family.THETA
+ && (!isThetaSimpleUnion(userOperands) ||
!isThetaSimpleUnion(materializedViewOperands))) {
+ return false;
+ }
+ return effectiveNominalEntries(family, userOperands) <=
effectiveNominalEntries(family, materializedViewOperands);
+ }
+
+ /// A theta sketch query with filter predicates and a post-aggregation (set)
expression cannot be
+ /// served by a collapsed MV sketch. This mirrors the aggregation function's
own threshold, which
+ /// treats fewer than 4 arguments (column, params, filter(s),
post-aggregation) as a simple union.
+ private static boolean isThetaSimpleUnion(@Nullable List<Expression>
operands) {
+ return operands == null || operands.size() <
THETA_POST_AGGREGATION_MIN_ARGUMENTS;
+ }
+
+ @Nullable
+ private Family family() {
+ if (_userFunctionName.contains("CPC")) {
+ return Family.CPC;
+ }
+ if (_userFunctionName.contains("THETA")) {
+ return Family.THETA;
+ }
+ if (_userFunctionName.contains("TUPLE")) {
+ return Family.TUPLE;
+ }
+ return null;
+ }
+
+ private static long effectiveNominalEntries(Family family, @Nullable
List<Expression> operands) {
+ Literal param = literalAt(operands, 1);
+ switch (family) {
+ case CPC:
+ if (param == null) {
+ return 1L << CommonConstants.Helix.DEFAULT_CPC_SKETCH_LGK;
+ }
+ if (param.isSetStringValue()) {
+ return nominalEntriesFromString(param.getStringValue(), 1L <<
CommonConstants.Helix.DEFAULT_CPC_SKETCH_LGK);
+ }
+ return 1L << numeric(param,
CommonConstants.Helix.DEFAULT_CPC_SKETCH_LGK);
+ case THETA:
+ if (param == null) {
+ return DEFAULT_THETA_NOMINAL_ENTRIES;
+ }
+ return param.isSetStringValue()
+ ? nominalEntriesFromString(param.getStringValue(),
DEFAULT_THETA_NOMINAL_ENTRIES)
+ : numeric(param, DEFAULT_THETA_NOMINAL_ENTRIES);
+ case TUPLE:
+ default:
+ long tupleDefault = 1L <<
CommonConstants.Helix.DEFAULT_TUPLE_SKETCH_LGK;
+ if (param == null) {
+ return tupleDefault;
+ }
+ return param.isSetStringValue()
+ ? nominalEntriesFromString(param.getStringValue(), tupleDefault)
+ : numeric(param, (int) tupleDefault);
+ }
+ }
+
+ @Nullable
+ private static Literal literalAt(@Nullable List<Expression> operands, int
index) {
+ if (operands == null || operands.size() <= index) {
+ return null;
+ }
+ Expression op = operands.get(index);
+ return op.getType() == ExpressionType.LITERAL ? op.getLiteral() : null;
+ }
+
+ private static long nominalEntriesFromString(String params, long
defaultValue) {
+ for (String pair : params.split(";")) {
+ String[] keyValue = pair.split("=", 2);
+ if (keyValue.length == 2 &&
keyValue[0].trim().equalsIgnoreCase(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES))
{
+ try {
+ return Long.parseLong(keyValue[1].trim());
Review Comment:
**[P1] Use runtime precedence for repeated precision parameters.** This
parser returns the first `nominalEntries` value, but the runtime parsers use
the last. A query with `'nominalEntries=4096;nominalEntries=65536'` therefore
passes against a default CPC MV even though execution requests lgK=16 and the
stored sketch has lgK=12. The targeted probe confirmed this rewrite is
accepted. Match runtime parsing semantics and add a strategy-level regression;
Theta and tuple have the same mismatch.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MaterializedViewClusterIntegrationTest.java:
##########
@@ -1270,6 +1396,165 @@ private void setupScalarGroupingMv()
/// Segment building helpers
/// -----------------------------------------------------------------------
+ /// Sketch MV: full coverage, groups by Carrier, stores a raw CPC and a raw
theta sketch of Origin.
+ /// The stored bytes are produced by the source table's own raw-sketch
aggregation, so a rewrite
+ /// that re-aggregates them returns exactly the base-table result.
+ private void setupSketchMv()
+ throws Exception {
+ Schema materializedViewSchema = new Schema.SchemaBuilder()
+ .setSchemaName(MATERIALIZED_VIEW_SKETCH_TABLE_NAME)
+ .addSingleValueDimension("Carrier", FieldSpec.DataType.STRING)
+ .addSingleValueDimension("raw_cpc_origin", FieldSpec.DataType.BYTES)
+ .addSingleValueDimension("raw_theta_origin", FieldSpec.DataType.BYTES)
Review Comment:
**[P3] Import DataType directly.** Import `FieldSpec.DataType` and use
`DataType.STRING` / `DataType.BYTES`, as required by the repository convention.
The same applies to the new schema declarations below.
--
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]
