Jackie-Jiang commented on code in PR #20002:
URL: https://github.com/apache/pinot/pull/20002#discussion_r4351609535
##########
pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/converter/RealtimeSegmentConverterTest.java:
##########
@@ -1641,6 +1641,121 @@ public void
testRangeIndexOnIngestionAggregatedNoDictColumn()
}
}
+ @DataProvider
+ public static Object[][] noDictSortedAggregationKeyParams() {
+ List<Object[]> params = new ArrayList<>();
+ for (DataType dataType : new DataType[]{DataType.INT, DataType.LONG,
DataType.STRING, DataType.BYTES}) {
+ params.add(new Object[]{dataType, false}); // row-major segment builder
+ params.add(new Object[]{dataType, true}); // column-major segment
builder
+ }
+ return params.toArray(new Object[0][]);
+ }
+
+ /// A sorted DIMENSION column configured as no-dictionary gets no inverted
index in the consuming segment, but
+ /// ingestion aggregation forces a dictionary on it because it is part of
the aggregation key. The sort step at seal
+ /// time used to take the inverted-index path because the column had a
dictionary, and failed with an NPE. This
+ /// test verifies that the segment builds, the rows aggregate, and the
sorted column stays sorted and raw.
+ @Test(dataProvider = "noDictSortedAggregationKeyParams")
+ public void testNoDictSortedColumnAsIngestionAggregationKey(DataType
dataType, boolean columnMajorSegmentBuilder)
+ throws Exception {
+ File tmpDir = new File(TMP_DIR,
+ "tmp_sorted_agg_" + dataType + "_" + columnMajorSegmentBuilder + "_" +
System.nanoTime());
+ String sortedColumn = "sorted_col";
+ List<AggregationConfig> aggregationConfigs =
+ List.of(new AggregationConfig(LONG_COLUMN4, "SUM(" + LONG_COLUMN4 +
")"));
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setAggregationConfigs(aggregationConfigs);
+ TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME)
+ .setTableName("testTable")
+ .setTimeColumnName(DATE_TIME_COLUMN)
+ .setSortedColumn(sortedColumn)
+ .setNoDictionaryColumns(List.of(sortedColumn, LONG_COLUMN4))
+ .setIngestionConfig(ingestionConfig)
+ .setColumnMajorSegmentBuilderEnabled(columnMajorSegmentBuilder)
+ .build();
+ Schema schema = new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .addSingleValueDimension(sortedColumn, dataType)
+ .addMetric(LONG_COLUMN4, DataType.LONG)
+ .addDateTime(DATE_TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .build();
+
+ String tableNameWithType = tableConfig.getTableName();
+ String segmentName = "testTable__0__0__123456";
+
+ RealtimeSegmentConfig realtimeSegmentConfig = new
RealtimeSegmentConfig.Builder(tableConfig, schema)
+ .setTableNameWithType(tableNameWithType)
+ .setSegmentName(segmentName)
+ .setStreamName(tableNameWithType)
+ .setSchema(schema)
+ .setTimeColumnName(DATE_TIME_COLUMN)
+ .setCapacity(1000)
+ .setIngestionAggregationConfigs(aggregationConfigs)
+ .setSegmentZKMetadata(getSegmentZKMetadata(segmentName))
+ .setOffHeap(true)
+ .setMemoryManager(new DirectMemoryManager(segmentName))
+ .setStatsHistory(RealtimeSegmentStatsHistory.deserializeFrom(new
File(tmpDir, "stats")))
+ .setConsumerDir(new File(tmpDir, "consumerDir").getAbsolutePath())
+ .build();
+
+ MutableSegmentImpl mutableSegment = new
MutableSegmentImpl(realtimeSegmentConfig, null);
+ try {
+ // Insert each key twice, in reverse sorted order, so that rows
aggregate and the conversion must reorder them.
+ // Row i carries metric value i + 1, so the aggregated value for key i
is 2 * (i + 1).
+ int numKeys = 10;
+ for (int round = 0; round < 2; round++) {
+ for (int i = numKeys - 1; i >= 0; i--) {
+ GenericRow row = new GenericRow();
+ row.putValue(sortedColumn, sortedValue(dataType, i, numKeys));
+ row.putValue(LONG_COLUMN4, (long) (i + 1));
+ row.putValue(DATE_TIME_COLUMN, 1697814309L);
Review Comment:
Optional coverage improvement: the constant time value means aggregation
leaves exactly one document per sorted value, so every counting-sort bucket has
size one. Could we vary the time or add another dimension so multiple documents
survive for the same sorted value, ideally with unequal bucket sizes? Keeping
duplicate complete keys would still exercise aggregation, while assertions on
the resulting keys, metrics, and original docId order within each sort value
would also cover bucket placement and stability in both builders.
--
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]
