findingrish commented on code in PR #16676:
URL: https://github.com/apache/druid/pull/16676#discussion_r1666652777
##########
server/src/main/java/org/apache/druid/segment/metadata/CoordinatorSegmentMetadataCache.java:
##########
@@ -419,6 +502,98 @@ private Set<SegmentId>
filterSegmentWithCachedSchema(Set<SegmentId> segmentIds)
return cachedSegments;
}
+ @VisibleForTesting
+ protected void coldDatasourceSchemaExec()
+ {
+ Collection<ImmutableDruidDataSource> immutableDataSources =
+
sqlSegmentsMetadataManager.getImmutableDataSourcesWithAllUsedSegments();
+
+ final Map<String, ColumnType> columnTypes = new LinkedHashMap<>();
+
+ Set<String> dataSources = new HashSet<>();
+
+ for (ImmutableDruidDataSource dataSource : immutableDataSources) {
+ String dataSourceName = dataSource.getName();
+ dataSources.add(dataSourceName);
+ Collection<DataSegment> dataSegments = dataSource.getSegments();
+
+ for (DataSegment segment : dataSegments) {
+ Integer replicationFactor =
segmentReplicationStatusManager.getReplicationFactor(segment.getId());
+ if (replicationFactor != null && replicationFactor != 0) {
+ continue;
+ }
+ Optional<SchemaPayloadPlus> optionalSchema =
segmentSchemaCache.getSchemaForSegment(segment.getId());
+ if (optionalSchema.isPresent()) {
+ RowSignature rowSignature =
optionalSchema.get().getSchemaPayload().getRowSignature();
+ for (String column : rowSignature.getColumnNames()) {
+ final ColumnType columnType =
+ rowSignature.getColumnType(column)
+ .orElseThrow(() -> new ISE("Encountered null type
for column [%s]", column));
+
+ columnTypes.compute(column, (c, existingType) ->
columnTypeMergePolicy.merge(existingType, columnType));
+ }
+ }
+ }
+
+ final RowSignature.Builder builder = RowSignature.builder();
+ columnTypes.forEach(builder::add);
+
+ RowSignature coldSignature = builder.build();
+
+ log.debug("[%s] signature from cold segments is [%s]", dataSourceName,
coldSignature);
+
+ coldSchemaTable.put(dataSourceName, new
DataSourceInformation(dataSourceName, coldSignature));
+
+ // update tables map with merged schema, if signature doesn't exist we
do not add entry in this table
+ // schema for entirely cold datasource is maintained separately
+ tables.computeIfPresent(
+ dataSourceName,
+ (ds, info) -> {
+ RowSignature mergedSignature =
mergeHotAndColdSchema(info.getRowSignature(), coldSignature);
+
+ if (!info.getRowSignature().equals(mergedSignature)) {
+ log.info(
+ "[%s] has new merged signature: %s. hot signature [%s], cold
signature [%s].",
+ ds, mergedSignature, info.getRowSignature(), coldSignature
+ );
+ } else {
+ log.debug("[%s] merged signature is unchanged.", ds);
+ }
+
+ return new DataSourceInformation(ds, mergedSignature);
+ }
+ );
+ }
+
+ // remove any stale datasource from the map
+ coldSchemaTable.keySet().retainAll(dataSources);
+ }
+
+ private RowSignature mergeHotAndColdSchema(RowSignature hot, RowSignature
cold)
Review Comment:
I can refactor this a bit to have a single method for merging the
RowSignature.
--
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]
