github-actions[bot] commented on code in PR #66345:
URL: https://github.com/apache/doris/pull/66345#discussion_r3701581054
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -431,7 +431,8 @@ public List<ConnectorScanRange> planScan(ConnectorSession
session, ConnectorScan
public long streamingSplitEstimate(ConnectorSession session,
ConnectorTableHandle handle,
Optional<ConnectorExpression> filter, boolean countPushdown) {
IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
- if (iceHandle.isSystemTable() || !sessionBool(session,
ENABLE_EXTERNAL_TABLE_BATCH_MODE, true)) {
+ if (iceHandle.isResolvedEmptySnapshot() || iceHandle.isSystemTable()
Review Comment:
[P1] Apply the empty-snapshot fence to streamed splits
This check runs from `computeBatchMode()` with the unpinned handle. If
query-begin resolved `-1` and a concurrent first append lands before
estimation, the refreshed table can cross the streaming threshold while this
condition is false. `startStreamingSplit()` then pins the empty handle, but
`streamSplits()` never rechecks the marker and treats `-1` as latest, exposing
the concurrent rows (and letting MERGE's anti-join read a different snapshot).
Please recheck `isResolvedEmptySnapshot()` in the post-pin streaming path, or
make the batch decision after pinning, and cover this first-append race.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java:
##########
@@ -265,24 +374,67 @@ public List<ConnectorWriteSortColumn>
getWriteSortColumns(ConnectorSession sessi
// unconditional setSortInfo inside the isSorted() branch even
when no identity column resolves.
return null;
}
- List<NestedField> columns = table.schema().columns();
+ Map<Integer, Integer> positionsByFieldId = new HashMap<>();
+ if (boundTargetColumns.isEmpty()) {
+ List<NestedField> currentColumns = table.schema().columns();
+ for (int i = 0; i < currentColumns.size(); i++) {
+ positionsByFieldId.put(currentColumns.get(i).fieldId(), i);
+ }
+ } else {
+ for (int i = 0; i < boundTargetColumns.size(); i++) {
+
positionsByFieldId.put(boundTargetColumns.get(i).getUniqueId(), i);
+ }
+ }
List<ConnectorWriteSortColumn> result = new ArrayList<>();
for (SortField sortField : sortOrder.fields()) {
if (!sortField.transform().isIdentity()) {
continue;
}
- for (int i = 0; i < columns.size(); i++) {
- if (columns.get(i).fieldId() == sortField.sourceId()) {
- result.add(new ConnectorWriteSortColumn(i,
- sortField.direction() == SortDirection.ASC,
- sortField.nullOrder() == NullOrder.NULLS_FIRST));
- break;
- }
+ Integer position = positionsByFieldId.get(sortField.sourceId());
+ if (position != null) {
+ // Resolve against the bound field id, never a newly refreshed
live ordinal; otherwise
+ // schema reorder can sort one output expression using another
column's ordering contract.
+ result.add(new ConnectorWriteSortColumn(position,
+ sortField.direction() == SortDirection.ASC,
+ sortField.nullOrder() == NullOrder.NULLS_FIRST));
}
}
return result;
}
+ @Override
+ public String getWriteMetadataIdentity(ConnectorSession session,
ConnectorTableHandle tableHandle) {
+ return writeMetadataIdentity(resolveTable(session,
(IcebergTableHandle) tableHandle));
+ }
+
+ private static String writeMetadataIdentity(Table table) {
Review Comment:
[P1] Include format version in the write-generation fence
This identity omits format version even though a v2-to-v3 upgrade changes
the physical REWRITE and UPDATE/MERGE schema by adding `_row_id` and
`_last_updated_sequence_number`. If binding/schema cache sees v2 but identity
capture and `beginWrite` see v3, sort/spec still match and column validation
filters the reserved fields, so the write is accepted; the v3 sink then
advertises two more columns than the bound output and BE rejects the arity.
Please fence format version (or the exact physical write-schema signature) and
add v2-bind/v3-plan tests for REWRITE and row-level writes.
--
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]
