github-advanced-security[bot] commented on code in PR #17726:
URL: https://github.com/apache/druid/pull/17726#discussion_r1954582087
##########
processing/src/main/java/org/apache/druid/query/JoinDataSource.java:
##########
@@ -419,164 +399,148 @@
'}';
}
- private DataSourceAnalysis getAnalysisForDataSource()
- {
- final Triple<DataSource, DimFilter, List<PreJoinableClause>> flattened =
flattenJoin(this);
- return new DataSourceAnalysis(flattened.first, null, flattened.second,
flattened.third);
- }
-
/**
* Creates a Function that maps base segments to {@link HashJoinSegment} if
needed (i.e. if the number of join
* clauses is > 0). If mapping is not needed, this method will return {@link
Function#identity()}.
- *
- * @param baseFilter Filter to apply before the join takes place
- * @param clauses Pre-joinable clauses
- * @param cpuTimeAccumulator An accumulator that we will add CPU nanos to;
this is part of the function to encourage
- * callers to remember to track metrics on CPU
time required for creation of Joinables
- * @param query The query that will be run on the mapped
segments. Usually this should be
- * {@code analysis.getBaseQuery().orElse(query)},
where "analysis" is a
- * {@link DataSourceAnalysis} and "query" is the
original
- * query from the end user.
+ * @param query
*/
- private Function<SegmentReference, SegmentReference>
createSegmentMapFunctionInternal(
- @Nullable final Filter baseFilter,
- final List<PreJoinableClause> clauses,
- final Query<?> query
- )
+ @Override
+ public Function<SegmentReference, SegmentReference>
createSegmentMapFunction(Query query)
{
- // compute column correlations here and RHS correlated values
+ DataSourceAnalysis safeAnalysis = getSafeAnalysisForDataSource();
+ List<PreJoinableClause> clauses = safeAnalysis.getPreJoinableClauses();
+ Filter baseFilter =
safeAnalysis.getJoinBaseTableFilter().map(Filters::toFilter).orElse(null);
+
if (clauses.isEmpty()) {
- return Function.identity();
- } else {
- final JoinableClauses joinableClauses = JoinableClauses.createClauses(
- clauses,
- joinableFactoryWrapper.getJoinableFactory()
+ throw DruidException.defensive("A JoinDataSource with no join clauses
should not be mapped.");
+ }
+ final JoinableClauses joinableClauses = JoinableClauses.createClauses(
+ clauses,
+ joinableFactoryWrapper.getJoinableFactory()
+ );
+ final JoinFilterRewriteConfig filterRewriteConfig =
JoinFilterRewriteConfig.forQuery(query);
+
+ // Pick off any join clauses that can be converted into filters.
+ final Set<String> requiredColumns = query.getRequiredColumns();
+ final Filter baseFilterToUse;
+ final List<JoinableClause> clausesToUse;
+
+ if (requiredColumns != null &&
filterRewriteConfig.isEnableRewriteJoinToFilter()) {
+ final Pair<List<Filter>, List<JoinableClause>> conversionResult =
JoinableFactoryWrapper.convertJoinsToFilters(
+ joinableClauses.getJoinableClauses(),
+ requiredColumns,
+
Ints.checkedCast(Math.min(filterRewriteConfig.getFilterRewriteMaxSize(),
Integer.MAX_VALUE))
);
- final JoinFilterRewriteConfig filterRewriteConfig =
JoinFilterRewriteConfig.forQuery(query);
-
- // Pick off any join clauses that can be converted into filters.
- final Set<String> requiredColumns = query.getRequiredColumns();
- final Filter baseFilterToUse;
- final List<JoinableClause> clausesToUse;
-
- if (requiredColumns != null &&
filterRewriteConfig.isEnableRewriteJoinToFilter()) {
- final Pair<List<Filter>, List<JoinableClause>> conversionResult =
JoinableFactoryWrapper.convertJoinsToFilters(
- joinableClauses.getJoinableClauses(),
- requiredColumns,
-
Ints.checkedCast(Math.min(filterRewriteConfig.getFilterRewriteMaxSize(),
Integer.MAX_VALUE))
- );
-
- baseFilterToUse =
- Filters.maybeAnd(
- Lists.newArrayList(
- Iterables.concat(
- Collections.singleton(baseFilter),
- conversionResult.lhs
- )
- )
- ).orElse(null);
- clausesToUse = conversionResult.rhs;
- } else {
- baseFilterToUse = baseFilter;
- clausesToUse = joinableClauses.getJoinableClauses();
- }
- // Analyze remaining join clauses to see if filters on them can be
pushed down.
- final JoinFilterPreAnalysis joinFilterPreAnalysis =
JoinFilterAnalyzer.computeJoinFilterPreAnalysis(
- new JoinFilterPreAnalysisKey(
- filterRewriteConfig,
- clausesToUse,
- query.getVirtualColumns(),
- Filters.maybeAnd(Arrays.asList(baseFilterToUse,
Filters.toFilter(query.getFilter())))
- .orElse(null)
+ baseFilterToUse = Filters.maybeAnd(
+ Lists.newArrayList(
+ Iterables.concat(
+ Collections.singleton(baseFilter),
+ conversionResult.lhs
+ )
)
- );
- final Function<SegmentReference, SegmentReference> baseMapFn;
- // A join data source is not concrete
- // And isConcrete() of an unnest datasource delegates to its base
- // Hence, in the case of a Join -> Unnest -> Join
- // if we just use isConcrete on the left
- // the segment map function for the unnest would never get called
- // This calls us to delegate to the segmentMapFunction of the left
- // only when it is not a JoinDataSource
- if (left instanceof JoinDataSource) {
- baseMapFn = Function.identity();
- } else {
- baseMapFn = left.createSegmentMapFunction(
- query
- );
- }
- return baseSegment ->
- new HashJoinSegment(
- baseMapFn.apply(baseSegment),
- baseFilterToUse,
- GuavaUtils.firstNonNull(clausesToUse, ImmutableList.of()),
- joinFilterPreAnalysis
- );
+ ).orElse(null);
+ clausesToUse = conversionResult.rhs;
+
+ } else {
+ baseFilterToUse = baseFilter;
+ clausesToUse = joinableClauses.getJoinableClauses();
}
+
+ // Analyze remaining join clauses to see if filters on them can be pushed
down.
+ final JoinFilterPreAnalysis joinFilterPreAnalysis =
JoinFilterAnalyzer.computeJoinFilterPreAnalysis(
+ new JoinFilterPreAnalysisKey(
+ filterRewriteConfig,
+ clausesToUse,
+ query.getVirtualColumns(),
+ Filters.maybeAnd(Arrays.asList(baseFilterToUse,
Filters.toFilter(query.getFilter())))
+ .orElse(null)
+ )
+ );
+ final Function<SegmentReference, SegmentReference> baseMapFn =
safeAnalysis.getBaseDataSource().createSegmentMapFunction(query);
+ return baseSegment -> newHashJoinSegment(
+ baseMapFn.apply(baseSegment),
+ baseFilterToUse,
+ clausesToUse,
+ joinFilterPreAnalysis
+ );
+ }
+
+ private SegmentReference newHashJoinSegment(
+ SegmentReference sourceSegment,
+ Filter baseFilterToUse,
+ List<JoinableClause> clausesToUse,
+ JoinFilterPreAnalysis joinFilterPreAnalysis)
+ {
+ if (clausesToUse.isEmpty() && baseFilterToUse == null) {
+ return sourceSegment;
+ }
+ return new HashJoinSegment(sourceSegment, baseFilterToUse, clausesToUse,
joinFilterPreAnalysis);
+ }
+
+
+ private DataSourceAnalysis getAnalysisForDataSource()
+ {
+ return flattenJoin(this, true);
+ }
+
+ public DataSourceAnalysis getSafeAnalysisForDataSource()
+ {
+ return flattenJoin(this, false);
}
/**
* Flatten a datasource into two parts: the left-hand side datasource (the
'base' datasource), and a list of join
* clauses, if any.
+ * @param b
Review Comment:
## Spurious Javadoc @param tags
@param tag "b" does not match any actual parameter of method "flattenJoin()".
[Show more
details](https://github.com/apache/druid/security/code-scanning/4889)
##########
processing/src/test/java/org/apache/druid/query/planning/DataSourceAnalysisTest.java:
##########
@@ -485,10 +485,11 @@
@Test
public void testJoinUnderTopLevelSubqueries()
{
+ final JoinDataSource joinDataSource;
Review Comment:
## Unread local variable
Variable 'JoinDataSource joinDataSource' is never read.
[Show more
details](https://github.com/apache/druid/security/code-scanning/4890)
--
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]
