bziobrowski commented on code in PR #15130:
URL: https://github.com/apache/pinot/pull/15130#discussion_r2011704995
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/DimensionTableDataManager.java:
##########
@@ -187,43 +221,79 @@ private DimensionTable createFastLookupDimensionTable() {
Preconditions.checkState(CollectionUtils.isNotEmpty(primaryKeyColumns),
"Primary key columns must be configured for dimension table: %s",
_tableNameWithType);
- Map<PrimaryKey, GenericRow> lookupTable = new HashMap<>();
List<SegmentDataManager> segmentDataManagers = acquireAllSegments();
try {
+ // count all documents to limit map re-sizings
+ int totalDocs = 0;
+ for (SegmentDataManager segmentManager : segmentDataManagers) {
+ IndexSegment indexSegment = segmentManager.getSegment();
+ totalDocs += indexSegment.getSegmentMetadata().getTotalDocs();
+ }
+
+ Object2ObjectOpenCustomHashMap<Object[], Object[]> lookupTable =
+ new Object2ObjectOpenCustomHashMap<>(totalDocs, HASH_STRATEGY);
+
+ List<String> valueColumns = getValueColumns(schema.getColumnNames(),
primaryKeyColumns);
+
for (SegmentDataManager segmentManager : segmentDataManagers) {
IndexSegment indexSegment = segmentManager.getSegment();
int numTotalDocs = indexSegment.getSegmentMetadata().getTotalDocs();
if (numTotalDocs > 0) {
try (PinotSegmentRecordReader recordReader = new
PinotSegmentRecordReader()) {
recordReader.init(indexSegment);
+
+ int[] pkIndexes =
recordReader.getIndexesForColumns(primaryKeyColumns);
+ int[] valIndexes = recordReader.getIndexesForColumns(valueColumns);
+
for (int i = 0; i < numTotalDocs; i++) {
if (_loadToken.get() != token) {
// Token changed during the loading, abort the loading
return null;
}
- GenericRow row = new GenericRow();
- recordReader.getRecord(i, row);
- GenericRow previousRow =
lookupTable.put(row.getPrimaryKey(primaryKeyColumns), row);
- if (_errorOnDuplicatePrimaryKey && previousRow != null) {
+
+ //TODO: consider inlining primary keys and values tables to
reduce memory overhead
+ Object[] primaryKey = recordReader.getRecordValues(i, pkIndexes);
+ Object[] values = recordReader.getRecordValues(i, valIndexes);
+
+ Object[] previousValue = lookupTable.put(primaryKey, values);
+ if (_errorOnDuplicatePrimaryKey && previousValue != null) {
throw new IllegalStateException(
"Caught exception while reading records from segment: " +
indexSegment.getSegmentName()
- + "primary key already exist for: " +
row.getPrimaryKey(primaryKeyColumns));
+ + "primary key already exist for: " +
toString(primaryKey));
}
}
} catch (Exception e) {
throw new RuntimeException(
- "Caught exception while reading records from segment: " +
indexSegment.getSegmentName());
+ "Caught exception while reading records from segment: " +
indexSegment.getSegmentName(), e);
}
}
}
- return new FastLookupDimensionTable(schema, primaryKeyColumns,
lookupTable);
+ return new FastLookupDimensionTable(schema, primaryKeyColumns,
valueColumns, lookupTable);
} finally {
for (SegmentDataManager segmentManager : segmentDataManagers) {
releaseSegment(segmentManager);
}
}
}
+ private static String toString(Object[] primaryKey) {
+ try {
+ return JsonUtils.objectToPrettyString(primaryKey);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static List<String> getValueColumns(NavigableSet<String>
columnNames, List<String> primaryKeyColumns) {
Review Comment:
It's part of init method so it'd end up being called in the manager class
anyway.
--
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-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
