github-actions[bot] commented on code in PR #69545:
URL: https://github.com/apache/doris/pull/69545#discussion_r4359737377
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java:
##########
@@ -204,13 +206,60 @@ public static Table applyOptions(Table table, Map<String,
String> options) {
}
// Apply the relation override first, then normalize each physical
planning branch. A
// single combined copy would flatten independent fallback values into
one sibling's cap.
- Table effectiveTable =
PaimonReaderOptions.runtimeSafeTable(table.copy(isolatedOptions));
+ Table effectiveTable = PaimonReaderOptions.runtimeSafeTable(
+ copyWithHistoricalSchema(table, isolatedOptions, options));
// Validate after every copy so relation options participate in the
documented
// relation > catalog > physical precedence before the effective value
is judged.
PaimonReaderOptions.validateEffectiveTable(effectiveTable);
return effectiveTable;
}
+ static Table copyWithHistoricalSchema(Table table, Map<String, String>
options) {
+ return copyWithHistoricalSchema(table, userOptions(options), options);
+ }
+
+ private static Table copyWithHistoricalSchema(
+ Table table, Map<String, String> dynamicOptions, Map<String,
String> provenance) {
+ if (!(table instanceof FileStoreTable) || table instanceof
ChainGroupReadTable
+ || PaimonTableDecorators.isChainTable(table)) {
+ // Chain groups have their own option filtering and are not
ordinary fallback readers.
+ return table.copy(dynamicOptions);
+ }
+ if (table instanceof FallbackReadFileStoreTable) {
+ FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable)
table;
+ // Each branch has independent snapshot IDs. Keep the SDK's
timestamp translation
+ // before restoring either schema; copying one historical schema
onto the pair is invalid.
+ return PaimonTableDecorators.rebuildFallbackPair(pair,
Review Comment:
[P2] Keep fallback priority consistent with the restored branch options
Suppose an old main schema has `scan.primary-branch=backup` and
`scan.manifest.parallelism=128`, then the current schema removes that primary
setting and uses `scan.fallback-branch=backup`. Copying the old snapshot here
preserves the current pair's `wrappedFirst=true`, but restores main options
containing only the old primary setting. On a 64-core FE, the subsequent
`runtimeSafeTable` caps parallelism and rebuilds this pair, re-inferring
`wrappedFirst=false` from those options; on a 128-core FE no cap is needed and
it stays true. With both branches populated in the same partition, the SDK
therefore selects backup rows on one FE and main rows on the other. Keep the
branch-policy options and retained pair priority consistent through restoration
and later rebuilds, and cover this policy change with overlapping-partition
tests at both capacities.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java:
##########
@@ -204,13 +206,60 @@ public static Table applyOptions(Table table, Map<String,
String> options) {
}
// Apply the relation override first, then normalize each physical
planning branch. A
// single combined copy would flatten independent fallback values into
one sibling's cap.
- Table effectiveTable =
PaimonReaderOptions.runtimeSafeTable(table.copy(isolatedOptions));
+ Table effectiveTable = PaimonReaderOptions.runtimeSafeTable(
+ copyWithHistoricalSchema(table, isolatedOptions, options));
// Validate after every copy so relation options participate in the
documented
// relation > catalog > physical precedence before the effective value
is judged.
PaimonReaderOptions.validateEffectiveTable(effectiveTable);
return effectiveTable;
}
+ static Table copyWithHistoricalSchema(Table table, Map<String, String>
options) {
+ return copyWithHistoricalSchema(table, userOptions(options), options);
+ }
+
+ private static Table copyWithHistoricalSchema(
+ Table table, Map<String, String> dynamicOptions, Map<String,
String> provenance) {
+ if (!(table instanceof FileStoreTable) || table instanceof
ChainGroupReadTable
+ || PaimonTableDecorators.isChainTable(table)) {
+ // Chain groups have their own option filtering and are not
ordinary fallback readers.
+ return table.copy(dynamicOptions);
+ }
+ if (table instanceof FallbackReadFileStoreTable) {
+ FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable)
table;
+ // Each branch has independent snapshot IDs. Keep the SDK's
timestamp translation
+ // before restoring either schema; copying one historical schema
onto the pair is invalid.
+ return PaimonTableDecorators.rebuildFallbackPair(pair,
+ (FileStoreTable) copyWithHistoricalSchema(pair.wrapped(),
dynamicOptions, provenance),
+ (FileStoreTable) copyWithHistoricalSchema(pair.other(),
+ PaimonTableDecorators.otherOptions(pair,
dynamicOptions), provenance));
+ }
+ FileStoreTable dataTable = (FileStoreTable) table;
+ Map<String, String> effectiveOptions = new HashMap<>(table.options());
+ dynamicOptions.forEach((key, value) -> {
+ if (value == null) {
+ effectiveOptions.remove(key);
+ } else {
+ effectiveOptions.put(key, value);
+ }
+ });
+ Optional<Snapshot> snapshot;
+ try {
+ snapshot =
TimeTravelUtil.tryTravelToSnapshot(Options.fromMap(effectiveOptions),
+ dataTable.snapshotManager(), dataTable.tagManager());
+ } catch (Exception e) {
+ // Match the SDK's schema-selection fallback (for example a
timestamp on an empty table).
+ // Its normal copy below still owns option validation and the
eventual scan error.
+ return table.copy(dynamicOptions);
+ }
+ if (snapshot.isPresent() && snapshot.get().schemaId() !=
dataTable.schema().id()) {
+ // Paimon overlays current options onto historical columns.
Restore the matching
+ // sequence/aggregation options first, or renamed fields fail
before scan planning.
+ dataTable = restoreBoundSchema(dataTable,
snapshot.get().schemaId(), provenance, "");
Review Comment:
[P2] Do not restore obsolete scan state with the historical schema
Consider a Paimon table created with `scan.mode=latest-full`, snapshot 1
written under that schema, and the option subsequently removed with
`SchemaChange.removeOption("scan.mode")`. After refreshing, `FOR VERSION AS OF
1` supplies only `scan.snapshot-id=1`. This restoration brings back the old
`latest-full` mode: `restoreBoundSchema` starts from historical options, and
its delta loop cannot remove a key absent from the current table. The final
`copy(dynamicOptions)` then fails SDK startup validation because `latest-full`
forbids `scan.snapshot-id`; the previous current-table copy succeeded. Preserve
the effective read-state family, including absent keys, while restoring
field-dependent historical options, and add a regression for this
removed-option history.
--
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]
