github-actions[bot] commented on code in PR #67725:
URL: https://github.com/apache/doris/pull/67725#discussion_r4398516109
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -489,6 +521,241 @@ public StatementContext createNextExecuteContext() {
return next;
}
+ public Map<String, DataMaskSpec> getOrEvaluateDataMasks(DataPolicyContext
policy,
+ java.util.function.Supplier<Map<String, DataMaskSpec>> evaluation)
{
+ return dataMaskEvaluations.computeIfAbsent(policy, key ->
Map.copyOf(evaluation.get()));
+ }
+
+ private static List<String> policySchemaFingerprint(List<Column> schema) {
+ return schema.stream().map(column ->
column.getName().toLowerCase(Locale.ROOT) + ":"
+ + DataType.fromCatalogType(column.getType()).toSql() + ":" +
column.isAllowNull())
+ .collect(Collectors.toList());
+ }
+
+ /** Establishes raw-value preservation from a complete, successful mask
evaluation before locks. */
+ public boolean proveInitialPartitionValues(UnboundRelation relation,
TableIf table) {
+ if (relation.getTableSnapshot().isPresent() ||
relation.getScanParams() != null
+ || !(table instanceof PluginDrivenExternalTable)
+ || !table.supportsExternalMetadataPreload()
+ || !((PluginDrivenExternalTable)
table).supportsConnectorPartitionPruning()
+ || ((PluginDrivenExternalTable) table).isView()
+ || table.getDatabase() == null ||
table.getDatabase().getCatalog() == null
+ || !(((PluginDrivenExternalTable) table).getCatalog()
instanceof PluginDrivenExternalCatalog)
+ || mvRefreshPredicates.isPresent()) {
+ return false;
+ }
+ requirePartitionPolicyPlanning();
+ PluginDrivenExternalTable external = (PluginDrivenExternalTable) table;
+ loadSnapshots(table, Optional.empty(), Optional.empty());
+ Optional<MvccSnapshot> pin = getSnapshot(table, Optional.empty(),
Optional.empty());
+ List<Column> schema = external.getFullSchema(pin);
+ if (schema == null || schema.isEmpty()) {
+ return false;
+ }
+ Optional<SchemaCacheValue> schemaValue =
external.getSchemaCacheValue(pin);
+ if (schemaValue.isEmpty() || !(schemaValue.get() instanceof
PluginDrivenSchemaCacheValue)) {
+ return false;
+ }
+ DataPolicyContext policy = capturePartitionPolicy(external, schema);
+ Map<String, DataMaskSpec> masks =
connectContext.getEnv().getAccessManager()
+ .evalDataMaskPolicies(policy, this);
+ partitionPolicyObservedTables.add(table.getId());
+ if (!masks.isEmpty()) {
+ return false;
+ }
+ partitionPolicyProofs.putIfAbsent(relation.getRelationId(),
+ new PartitionPolicyProof(external, policy, pin,
policySchemaFingerprint(schema),
+
policySchemaFingerprint(external.getPartitionColumns(pin)),
+ ((PluginDrivenSchemaCacheValue)
schemaValue.get()).getPartitionColumnRemoteNames()));
+ return true;
+ }
+
+ private DataPolicyContext capturePartitionPolicy(PluginDrivenExternalTable
table, List<Column> schema) {
+ Set<String> columns =
schema.stream().map(Column::getName).collect(Collectors.toCollection(LinkedHashSet::new));
+ return capturePartitionPolicyColumns(table, columns);
+ }
+
+ private DataPolicyContext
capturePartitionPolicyColumns(PluginDrivenExternalTable table, Set<String>
columns) {
+ return
connectContext.getEnv().getAccessManager().captureDataPolicyContext(
+ connectContext.getCurrentUserIdentity(),
table.getDatabase().getCatalog().getName(),
+ table.getDatabase().getFullName(), table.getName(), columns,
this);
+ }
+
+ public boolean hasPartitionPolicyObservation(TableIf table) {
+ return partitionPolicyObservedTables.contains(table.getId());
+ }
+
+ /** Refuses a bound relation whose actual policy scope or typed output no
longer justifies its subset. */
+ public DataPolicyContext validatePartitionPolicy(LogicalPlan relation) {
+ requirePartitionPolicyPlanning();
+ if (!(relation instanceof
org.apache.doris.nereids.trees.plans.logical.LogicalFileScan)) {
+ return null;
+ }
+ org.apache.doris.nereids.trees.plans.logical.LogicalFileScan scan =
+ (org.apache.doris.nereids.trees.plans.logical.LogicalFileScan)
relation;
+ PartitionPolicyProof proof =
partitionPolicyProofs.get(scan.getPreloadRelationId());
+ if (proof == null || !usedPartitionPolicyProofs.contains(proof)) {
+ return null;
+ }
+ Set<String> columns =
scan.getOutput().stream().map(Slot::getName).collect(Collectors.toSet());
+ List<String> output = scan.getOutput().stream().map(slot ->
slot.getName().toLowerCase(Locale.ROOT)
+ + ":" + slot.getDataType().toSql() + ":" +
slot.nullable()).collect(Collectors.toList());
+ try {
+ DataPolicyContext actual =
capturePartitionPolicyColumns(proof.table, columns);
+ if (proof.table != scan.getTable() || !proofMatches(proof, actual,
false) || !proof.schema.equals(output)) {
+ throw new AnalysisException("Partition policy proof changed
before binding " + proof.table.getName());
+ }
+ return actual;
+ } catch (RuntimeException failure) {
+ invalidatePartitionPolicy(false);
+ throw failure;
+ }
+ }
+
+ /** Records the raw row-policy observation actually expanded into this
relation's plan. */
+ public void recordPartitionRowPolicies(LogicalPlan relation,
List<RowFilterSpec> policies) {
+ if (relation instanceof
org.apache.doris.nereids.trees.plans.logical.LogicalFileScan) {
+ PartitionPolicyProof proof = partitionPolicyProofs.get(
+
((org.apache.doris.nereids.trees.plans.logical.LogicalFileScan)
relation).getPreloadRelationId());
+ if (proof != null) {
+ List<RowFilterSpec> previous =
partitionProofRowPolicies.putIfAbsent(proof, List.copyOf(policies));
+ if (previous != null && !previous.equals(policies)) {
+ invalidatePartitionPolicy(false);
+ throw new AnalysisException("Row policy changed between
aliases of " + proof.table.getName());
+ }
+ }
+ }
+ }
+
+ private boolean proofMatches(PartitionPolicyProof proof, DataPolicyContext
actual, boolean retry) {
+ DataPolicyContext expected =
executionPolicyContexts.getOrDefault(proof, proof.policy);
+ return proof.table.getDatabase() != null &&
proof.table.getDatabase().getCatalog() != null
+ && ((PluginDrivenExternalCatalog)
proof.table.getCatalog()).getConnector() == proof.connector
+ && proof.pin.orElse(null) == getSnapshot(proof.table,
Optional.empty(), Optional.empty()).orElse(null)
+ && proofSchemaMatches(proof)
+ && (retry ? expected.samePolicyScope(actual) :
expected.equals(actual));
+ }
+
+ private boolean proofSchemaMatches(PartitionPolicyProof proof) {
+ Optional<SchemaCacheValue> schema =
proof.table.getSchemaCacheValue(proof.pin);
+ if (schema.isEmpty() || !(schema.get() instanceof
PluginDrivenSchemaCacheValue)) {
+ return false;
+ }
+ PluginDrivenSchemaCacheValue actual = (PluginDrivenSchemaCacheValue)
schema.get();
+ return proof.schema.equals(policySchemaFingerprint(actual.getSchema()))
+ &&
proof.partitions.equals(policySchemaFingerprint(actual.getPartitionColumns()))
+ &&
proof.remotePartitions.equals(actual.getPartitionColumnRemoteNames());
+ }
+
+ private void requirePartitionPolicyPlanning() {
+ if (partitionPolicyPlanningInvalidated) {
+ throw new AnalysisException("Partition policy planning state was
invalidated; replan in a fresh statement");
+ }
+ }
+
+ private void invalidatePartitionPolicy(boolean retry) {
+ partitionPolicyEpoch++;
+ dataMaskEvaluations.clear();
+ partitionPolicyObservedTables.clear();
+ boolean used = !usedPartitionPolicyProofs.isEmpty();
+ for (Map.Entry<RelationId, PartitionPolicyProof> entry :
partitionPolicyProofs.entrySet()) {
+ ExternalTablePreloadInfo info =
externalTablePreloadInfos.get(entry.getValue().table.getId());
+ if (info != null) {
+ info.getInitialScanPredicates().remove(entry.getKey());
+ info.markUnfilteredLatestRelation();
+ info.clearPartitionViews();
+ }
+ }
+ partitionPolicyPlanningInvalidated |= used;
Review Comment:
[P2] Allow actual replans after resetting the policy proof
For an ordinary user with no masks, a mixed query with `Filter(h.p = 'x') ->
CrossJoin(internal_i, HiveScan(h))` uses a selective preload and puts its proof
in `usedPartitionPolicyProofs`. If execution then reports a cloud replan error
such as `E-230`, `StmtExecutor.executeWithReplanRetries` calls
`resetConnectorStatementScope()`, which reaches this assignment, and creates a
new `NereidsPlanner` with the same `StatementContext`. The next
collect/preload/lock therefore always throws `Partition policy planning state
was invalidated; replan in a fresh statement`, even if the transient error is
gone. The `InsertIntoTableCommand.initPlan` retry loop has the same failure
when an INSERT SELECT target changes identity/schema after its first plan. Keep
the old plan revoked while giving actual replan callers fresh planning/proof
state, and cover an outer replan retry; the new RPC-retry tests only redispatch
the existing plan.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -1511,6 +1799,249 @@ public Collection<ExternalTablePreloadInfo>
getExternalTablePreloadInfos() {
return
Collections.unmodifiableCollection(externalTablePreloadInfos.values());
}
+ /** Marks collected view relations whose ids will be replaced when the
definition is bound again. */
+ public void markReparsedViewRelationsForPreload(Set<RelationId>
relationIds) {
+ externalTablePreloadInfos.values().forEach(info ->
+ relationIds.forEach(info::markReparsedScanRelation));
+ }
+
+ public boolean shouldSkipUnpreloadedScanPartitionPruning() {
+ return skipUnpreloadedScanPartitionPruning;
+ }
+
+ /**
+ * The preload record of one external table, or empty when the table was
never registered for preload.
+ * Used by lock-sensitive consumers to reuse metadata the pre-lock preload
pass already materialized.
+ *
+ * @param tableId the table's id, as used by {@link
#registerExternalTableForPreload}
+ * @return ExternalTablePreloadInfo
+ */
+ public Optional<ExternalTablePreloadInfo> getExternalTablePreloadInfo(long
tableId) {
+ return Optional.ofNullable(externalTablePreloadInfos.get(tableId));
+ }
+
+ /**
+ * Materializes every deferred connector partition view that can be
consumed while the internal table read
+ * locks are held. This step is unconditional for physical finalization:
it exists so the default
+ * configuration gets the lock scope, not only the opt-in {@code
enable_preload_external_metadata} pass.
+ *
+ * <p>WHY it is needed at all: {@code QueryPartitionCollector} runs from
+ * {@code InitMaterializationContextHook.afterRewrite}, i.e. while {@link
#lock()} is held, and an unfiltered
+ * connector-pruning file scan is still {@code DEFERRED} at that point.
Even with both MV rewrite switches
+ * off, {@code PluginDrivenScanNode.doFinalize} still materializes a
no-filter deferred scan before the
+ * planner releases the internal read locks. Materializing the same view
here moves both consumers outside
+ * the lock window.</p>
+ *
+ * <p>Plan-only EXPLAIN still runs logical partition pruning, so
relation-scoped filters must be resolved
+ * before locking even when both MV rewrite switches are off. Only the
unfiltered full-view warmup can
+ * be skipped in that configuration. The MV gate is the OR of the query
and DML rewrite switches because
+ * either hook can consume a deferred view.</p>
+ */
+ public void preloadDeferredScanPartitionViewsBeforeLock(boolean
willFinalizePhysicalPlan) {
+ requirePartitionPolicyPlanning();
+ ConnectContext connectContext = getConnectContext();
+ if (connectContext == null || connectContext.getSessionVariable() ==
null) {
+ return;
+ }
+ boolean mvCollectorWillRun =
connectContext.getSessionVariable().isEnableMaterializedViewRewrite()
+ ||
connectContext.getSessionVariable().isEnableDmlMaterializedViewRewrite();
+ boolean needsUnfilteredScanView = willFinalizePhysicalPlan ||
mvCollectorWillRun;
+ skipUnpreloadedScanPartitionPruning = !needsUnfilteredScanView;
+ boolean hasPlanReadLockTable = hasAnyPlanReadLockTable();
+ Set<BaseTableInfo> candidateMtmvPctTables = mvCollectorWillRun
+ ? collectCandidateMtmvPctTables() : Collections.emptySet();
+ for (ExternalTablePreloadInfo preloadInfo :
externalTablePreloadInfos.values()) {
+ if (hasPlanReadLockTable) {
+ preloadFilteredScanPartitionViews(preloadInfo);
+ if (needsUnfilteredScanView) {
+ preloadDeferredScanPartitionView(preloadInfo);
+ }
+ }
+ preloadMtmvPartitionView(preloadInfo, candidateMtmvPctTables);
+ }
+ }
+
+ /** Resolve relation-scoped filters before internal locks, never sharing
one alias's subset with another. */
+ private void preloadFilteredScanPartitionViews(ExternalTablePreloadInfo
info) {
+ ExternalTable table = info.getTable();
+ if (!info.shouldPreloadLatestSnapshot() || !(table instanceof
PluginDrivenExternalTable)
+ || !((PluginDrivenExternalTable)
table).supportsConnectorPartitionPruning()) {
+ return;
+ }
+ loadSnapshots(table, Optional.empty(), Optional.empty());
+ Optional<MvccSnapshot> snapshot = getSnapshot(table);
+ List<Column> columns = table.getPartitionColumns(snapshot);
+ for (Map.Entry<RelationId, Expression> entry :
info.getInitialScanPredicates().entrySet()) {
+ if (info.hasFilteredScanView(entry.getKey())) {
+ continue;
+ }
+ PartitionPolicyProof proof =
partitionPolicyProofs.get(entry.getKey());
+ if (proof != null && !proofMatches(proof,
+ capturePartitionPolicy(proof.table,
proof.table.getFullSchema(snapshot)), false)) {
+ invalidatePartitionPolicy(false);
+ throw new AnalysisException("Partition policy changed before
selective preload");
+ }
+ // Only bind declared partition columns here. Full analysis,
coercion, privileges and residual
+ // evaluation remain in their normal stages; unsupported forms use
the preloaded full view.
+ Expression bound = entry.getValue().rewriteUp(expression -> {
+ if (!(expression instanceof UnboundSlot)) {
+ return expression;
+ }
+ List<String> names = ((UnboundSlot) expression).getNameParts();
+ String name = names.get(names.size() - 1);
+ return columns.stream().filter(column ->
column.getName().equalsIgnoreCase(name))
+ .findFirst().<Expression>map(column -> new
SlotReference(column.getName(),
+ DataType.fromCatalogType(column.getType()),
column.isAllowNull()))
+ .orElse(expression);
+ });
+ List<Expression> safeConjuncts =
ExpressionUtils.extractConjunction(bound).stream()
+
.filter(StatementContext::isSafeInitialPartitionPredicate).collect(Collectors.toList());
+ ConnectorExpression predicate = safeConjuncts.isEmpty() ? null
+ :
NereidsToConnectorExpressionConverter.convert(ExpressionUtils.and(safeConjuncts));
+ Optional<PluginDrivenExternalTable.ConnectorFilteredPartitionView>
view = predicate == null
+ ? Optional.empty()
+ : ((PluginDrivenExternalTable)
table).applyPartitionFilterForScan(snapshot, predicate);
+ info.setFilteredScanView(entry.getKey(), view, predicate);
+ if (proof != null && view.isPresent()) {
+ usedPartitionPolicyProofs.add(proof);
+ }
+ if (!view.isPresent() && !info.hasScanPartitionView()) {
+ info.setScanPartitionView(((PluginDrivenExternalTable)
table).getNameToPartitionItemsForScan(snapshot));
+ }
+ }
+ }
+
+ /** Only prefilter typed equality/IN whose later coercion cannot change
partition-value semantics. */
+ private static boolean isSafeInitialPartitionPredicate(Expression
expression) {
+ if (expression instanceof EqualTo) {
+ Expression left = expression.child(0);
+ Expression right = expression.child(1);
+ return left instanceof SlotReference && right instanceof Literal
+ && compatibleInitialPartitionTypes(left.getDataType(),
right.getDataType())
+ || right instanceof SlotReference && left instanceof
Literal
+ && compatibleInitialPartitionTypes(right.getDataType(),
left.getDataType());
+ }
+ if (expression instanceof InPredicate && expression.child(0)
instanceof SlotReference) {
+ DataType type = expression.child(0).getDataType();
+ return expression.children().subList(1,
expression.arity()).stream()
+ .allMatch(child -> child instanceof Literal
+ && compatibleInitialPartitionTypes(type,
child.getDataType()));
+ }
+ return false;
+ }
+
+ private static boolean compatibleInitialPartitionTypes(DataType column,
DataType literal) {
+ // In particular, VARCHAR p = 1 is a numeric comparison, not textual p
= '1': prefiltering it as
+ // strings could drop p=01. Let full analysis/coercion handle mixed
families on the warmed full view.
+ return column.equals(literal) || column.isIntegralType() &&
literal.isIntegralType()
+ || column.isStringLikeType() && literal.isStringLikeType();
+ }
+
+ /**
+ * Takes a stable copy of the candidate MVs' PCT table identities without
doing connector I/O while an MV lock
+ * is held. The full external views are materialized only after every MV
lock has been released.
+ */
+ private Set<BaseTableInfo> collectCandidateMtmvPctTables() {
+ Set<BaseTableInfo> pctTables = new HashSet<>();
+ for (MTMV mtmv : candidateMTMVs) {
+ mtmv.readMvLock();
+ try {
+ mtmv.getMvPartitionInfo().getPctInfos().forEach(
+ pctInfo -> pctTables.add(pctInfo.getTableInfo()));
+ } finally {
+ mtmv.readMvUnlock();
+ }
+ }
+ return pctTables;
+ }
+
+ /**
+ * Materializes the complete latest partition view needed by an async-MV
candidate's partition mapping. This
+ * is deliberately separate from the scan view: a filtered scan must
retain connector predicate pruning, while
+ * {@code MTMV.calculatePartitionMappings} still needs the full partition
universe for union compensation.
+ */
+ private void preloadMtmvPartitionView(ExternalTablePreloadInfo preloadInfo,
+ Set<BaseTableInfo> candidateMtmvPctTables) {
+ if (preloadInfo.hasMtmvPartitionView() ||
!preloadInfo.shouldPreloadLatestSnapshot()
+ || candidateMtmvPctTables.isEmpty()) {
+ return;
+ }
+ ExternalTable table = preloadInfo.getTable();
+ if (!(table instanceof PluginDrivenExternalTable)
+ || !((PluginDrivenExternalTable)
table).supportsConnectorPartitionPruning()
+ || !candidateMtmvPctTables.contains(new BaseTableInfo(table)))
{
+ return;
+ }
+ if (preloadInfo.hasScanPartitionView()) {
+
preloadInfo.setMtmvPartitionView(preloadInfo.getScanPartitionView());
+ return;
+ }
+ preloadInfo.setMtmvPartitionView(
Review Comment:
[P2] Skip full mapping warmup for ineligible external MVs
With the default
`materialized_view_rewrite_enable_contain_external_table=false`, an existing
partitioned Hive MV is collected as a candidate, but
`InitMaterializationContextHook.getAvailableMTMVs` discards it before partition
validation. A selective `Filter(h.p = 'x') -> HiveScan(h)` nevertheless reaches
this new mapping-view warmup because `collectCandidateMtmvPctTables` includes
every candidate. It lists/builds all N partitions after selective preload, even
though no consumer will inspect this full map; this also happens when the query
names only the Hive base. Thus merely having an ineligible MV restores the O(N)
planning cost the HMS filter is meant to avoid and can make the query fail on
an unnecessary full listing. Apply the same eligibility gates before collecting
tables to warm, and cover a selective query with a partitioned Hive MV and this
switch disabled.
--
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]
