github-actions[bot] commented on code in PR #67904:
URL: https://github.com/apache/doris/pull/67904#discussion_r4011946978
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java:
##########
@@ -343,10 +344,43 @@ Table resolveTable(PaimonTableHandle paimonHandle) {
*/
Table resolveScanTable(PaimonTableHandle paimonHandle) {
Table table = resolveTable(paimonHandle);
- Map<String, String> scanOptions = paimonHandle.getScanOptions();
+ return withBoundSchemaAuthentication(paimonHandle, () ->
applyScanOptions(paimonHandle, table));
+ }
+
+ private <T> T withBoundSchemaAuthentication(PaimonTableHandle handle,
Supplier<T> action) {
+ if (context == null ||
!PaimonScanParams.preservesBoundSchema(handle.getScanOptions())) {
+ return action.get();
+ }
+ // Restoring a bound schema can read FileIO after table resolution has
left the authenticated scope.
+ try {
+ return context.executeAuthenticated(action::get);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to restore Paimon statement
schema", e);
+ }
+ }
+
+ private Map<String, String> effectiveScanOptions(PaimonTableHandle handle)
{
+ Map<String, String> options = handle.getScanOptions();
+ return PaimonScanParams.preservesBoundSchema(options)
+ ? PaimonScanParams.withCatalogOptions(options,
PaimonTableOptions.extract(catalogProps.getRaw()))
Review Comment:
[P1] Please preserve the replay-compatible option set at this new
restoration boundary. `PaimonConnector` deliberately binds
`extractCompatible(properties)` so a catalog persisted before the current
reader-option allowlist can still restart while unknown or malformed
`paimon.table-option.*` values are ignored. Every ordinary latest pin now
reaches this branch, but reparsing `catalogProps.getRaw()` with strict
`extract` makes that catalog start successfully and then fail every query
during scan planning. Please reuse the already-filtered connector options here
(or call `extractCompatible`) and add a restart/replay case for a legacy value.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java:
##########
@@ -196,14 +201,79 @@ public static FileStoreTable
applyOptionsWithoutTimeTravel(
.filter(key -> !tableOptions.containsKey(key))
.forEach(key -> isolatedOptions.put(key, null));
}
- // The statement fence already selected the schema generation.
Preserve that generation
- // while carrying only the resolved read selector and execution
options into this copy.
+ String schemaId = options.get(BOUND_SCHEMA_ID);
+ PaimonSchemaPin.validate(table, options);
+ if (schemaId != null) {
+ table = restoreBoundSchema(table, Long.parseLong(schemaId),
options, "");
+ }
FileStoreTable effectiveTable = (FileStoreTable)
PaimonReaderOptions.runtimeSafeTable(
- table.copyWithoutTimeTravel(isolatedOptions));
+ copyWithPinnedFallback(table, isolatedOptions, options, ""));
PaimonReaderOptions.validateEffectiveTable(effectiveTable);
return effectiveTable;
}
+ private static FileStoreTable copyWithPinnedFallback(FileStoreTable table,
Map<String, String> dynamicOptions,
+ Map<String, String> coordinates, String path) {
+ if (table instanceof FallbackReadFileStoreTable) {
+ FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable)
table;
+ String fallbackPath = path + "fallback.";
+ String snapshotId =
PaimonSchemaPin.fallbackSnapshotId(coordinates, fallbackPath);
+ if (snapshotId != null) {
+ Map<String, String> fallbackOptions = new
HashMap<>(dynamicOptions);
+ // Keep branch policy, but never retranslate the main fence
against a later fallback history.
+ fallbackOptions.remove(CoreOptions.BUCKET.key());
+ fallbackOptions.put(CoreOptions.BRANCH.key(),
pair.fallback().coreOptions().branch());
+ fallbackOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(),
snapshotId);
+ return new FallbackReadFileStoreTable(
+ copyWithPinnedFallback(pair.wrapped(), dynamicOptions,
coordinates, path),
+ copyWithPinnedFallback(pair.fallback(),
fallbackOptions, coordinates, fallbackPath));
+ }
+ }
+ if (table instanceof DelegatedFileStoreTable && !(table instanceof
FallbackReadFileStoreTable)) {
+ return PaimonTableDecorators.replaceWrapped(table,
copyWithPinnedFallback(
+ ((DelegatedFileStoreTable) table).wrapped(),
dynamicOptions, coordinates, path));
+ }
+ return table.copyWithoutTimeTravel(dynamicOptions);
+ }
+
+ private static FileStoreTable restoreBoundSchema(
+ FileStoreTable table, long schemaId, Map<String, String> options,
String path) {
+ if (table instanceof FallbackReadFileStoreTable) {
+ FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable)
table;
+ // Each branch has its own schema history. Restore both children
at their captured
+ // coordinates, preserving compatibility without broadcasting the
main schema ID.
+ String fallbackPath = path + "fallback.";
+ long fallbackId = PaimonSchemaPin.fallbackSchemaId(options,
fallbackPath,
+ () ->
pair.fallback().schemaManager().latest().orElseThrow(IllegalStateException::new).id());
+ return new
FallbackReadFileStoreTable(restoreBoundSchema(pair.wrapped(), schemaId,
options, path),
+ restoreBoundSchema(pair.fallback(), fallbackId, options,
fallbackPath));
+ }
+ if (table instanceof DelegatedFileStoreTable) {
+ FileStoreTable wrapped = ((DelegatedFileStoreTable)
table).wrapped();
+ return PaimonTableDecorators.replaceWrapped(table,
restoreBoundSchema(wrapped, schemaId, options, path));
+ }
+ if (table.schema().id() == schemaId) {
Review Comment:
[P1] Verify the retained table object's generation before either restoration
branch. A DROP/CREATE can happen after `resolveTable` returns A but before
`statementPin` reads the live schema, so the pin/digests describe B while
`withScanOptions` keeps A. If IDs match, this returns A; if they differ, the
code below compares A's embedded options with B's unrelated history and replays
false dynamic overrides. Either path can leave B-bound FE metadata with
A-derived native/JNI state. The added race test recreates after `latestSchema`,
covering the opposite ordering. Please compare the pinned digest with
`table.schema()` for each delegated/fallback child, or reload/fail before both
branches, and test the reversed ordering with equal and unequal IDs.
--
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]
