Gabriel39 commented on code in PR #65851:
URL: https://github.com/apache/doris/pull/65851#discussion_r3694578458
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -1905,6 +2274,76 @@ public void close() throws IOException {
}
}
+ /** Identity of one predicate/ref/rewrite-scoped scan within a statement.
*/
+ private static final class ScanPlanKey {
+ private final IcebergTableHandle handle;
+ private final String filter;
+
+ private ScanPlanKey(IcebergTableHandle handle,
Optional<ConnectorExpression> filter) {
+ this.handle = handle;
+ this.filter = filter.map(Object::toString).orElse("");
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) {
+ return true;
+ }
+ if (!(other instanceof ScanPlanKey)) {
+ return false;
+ }
+ ScanPlanKey that = (ScanPlanKey) other;
+ return handle.equals(that.handle) && filter.equals(that.filter);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(handle, filter);
+ }
+ }
+
+ /**
+ * Materialized byte-split tasks discovered while building the schema
carrier. Equality-delete field IDs
+ * must come from the exact predicate-filtered tasks that execution will
read; keeping the tasks avoids a
+ * second metadata plan observing a different snapshot or delete set.
+ */
+ private static final class PreplannedSplitPlan {
+ private final List<FileScanTask> tasks;
+ private final long targetSplitSize;
+
+ private PreplannedSplitPlan(List<FileScanTask> tasks, long
targetSplitSize) {
+ this.tasks = tasks;
+ this.targetSplitSize = targetSplitSize;
+ }
+
+ private SplitPlan asSplitPlan() {
+ return new SplitPlan(CloseableIterable.withNoopClose(tasks),
targetSplitSize);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Map<ScanPlanKey, PreplannedSplitPlan>
preplannedScans(ConnectorSession session) {
+ if (session == null || session.getStatementScope() ==
ConnectorStatementScope.NONE) {
+ return new ConcurrentHashMap<>();
+ }
+ String key = PREPLANNED_SCAN_NAMESPACE + ":" + session.getCatalogId()
+ + ":" + session.getQueryId();
+ return session.getStatementScope().computeIfAbsent(key,
ConcurrentHashMap::new);
+ }
+
+ private PreplannedSplitPlan preplanScan(TableScan scan, ConnectorSession
session, Table table,
+ Optional<ConnectorExpression> filter) {
+ List<FileScanTask> tasks = new ArrayList<>();
Review Comment:
[P1] Keep equality-delete preflight bounded and preserve batch planning
`planFileScanTask()` returns byte-split tasks, and this method exhausts the
iterable into an `ArrayList` before any scan range can be dispatched. With a
small `file_split_size`, a table can produce millions of entries even when its
data-file count is much smaller. The cached plan also makes
`streamingSplitEstimate()` return `-1`, so every snapshot whose equality-delete
summary is missing or nonzero synchronously materializes the complete task set
on the FE and disables asynchronous batch dispatch. This creates unbounded
planning latency and FE heap growth, including an OOM risk, for exactly the
large scans that batch mode is intended to protect.
Please derive the equality-field carrier without retaining every byte-split
task, or introduce a bounded/lazy handoff that preserves the exact task/delete
set while allowing batch dispatch. Add a large-task regression test that proves
planning does not fully consume the iterator before dispatch and that retained
FE state remains bounded.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -246,6 +246,9 @@ public enum TableFrom {
private final Map<List<String>, Pair<String, Map<String, String>>>
viewInfos = Maps.newHashMap();
// save insert into schema to avoid schema changed between two read locks
private final List<Column> insertTargetSchema = new ArrayList<>();
+ // Request-scoped connector writer schemas. These are resolved before
DEFAULT/omitted-column expansion and
+ // are intentionally separate from the cached table schema used by
DESCRIBE/SHOW CREATE.
+ private final Map<Long, List<Column>> connectorWriteSchemas = new
HashMap<>();
Review Comment:
[P1] Do not retain a connector write schema across statement executions
Prepared `EXECUTE` reuses the same `StatementContext`, but
`resetConnectorStatementScope()` only closes and nulls
`connectorStatementScope`; this map survives.
`ConnectorWriteSchemaUtils.pinAndGet()` then sees the old entry and skips
`resolveWriteColumns()`. If an Iceberg column's write default changes from D1
to D2 between two executions of a prepared `UPDATE`, the analyzer can still
expand `DEFAULT(col)` with D1 while `IcebergWritePlanProvider` creates a fresh
D2 write context. The current-schema fence validates that fresh context against
the table, but it does not prove that the already-analyzed default literal came
from the same schema, so a type-compatible change can commit the stale D1 value
silently.
Please clear this state together with the connector statement scope, move it
into that scope, or key it by execution identity and validate analysis/sink
schema consistency. Add a prepared row-level DML test that executes once with
D1, changes the Iceberg write default, executes again, and verifies D2 is
written.
--
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]
