github-actions[bot] commented on code in PR #66297:
URL: https://github.com/apache/doris/pull/66297#discussion_r3698680226
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java:
##########
@@ -349,6 +370,58 @@ public MvccSnapshot loadSnapshot(Optional<TableSnapshot>
tableSnapshot, Optional
return new
PaimonMvccSnapshot(getPaimonSnapshotCacheValue(tableSnapshot, scanParams));
}
+ @Override
+ public MvccSnapshot loadLatestSnapshotFence() {
+ return new
PaimonMvccSnapshot(PaimonUtils.loadLatestSnapshotFence(this));
+ }
+
+ @Override
+ public boolean requiresLatestSnapshotFence(
+ Optional<TableSnapshot> tableSnapshot, Optional<TableScanParams>
scanParams) {
+ return !tableSnapshot.isPresent()
+ && scanParams.isPresent()
+ && scanParams.get().isOptions()
+ &&
PaimonScanParams.usesStatementSnapshot(scanParams.get().getMapParams());
+ }
+
+ @Override
+ public MvccSnapshot loadSnapshot(
+ Optional<TableSnapshot> tableSnapshot,
+ Optional<TableScanParams> scanParams,
+ Optional<MvccSnapshot> latestSnapshotFence) {
+ if (latestSnapshotFence.isPresent() && !tableSnapshot.isPresent() &&
!scanParams.isPresent()) {
+ PaimonSnapshot fence = ((PaimonMvccSnapshot)
latestSnapshotFence.get())
+ .getSnapshotCacheValue().getSnapshot();
+ return new
PaimonMvccSnapshot(PaimonUtils.loadSnapshotAtFence(this, fence));
+ }
+ if (!latestSnapshotFence.isPresent()
+ || !requiresLatestSnapshotFence(tableSnapshot, scanParams)) {
+ return loadSnapshot(tableSnapshot, scanParams);
+ }
+ PaimonMvccSnapshot fence = (PaimonMvccSnapshot)
latestSnapshotFence.get();
+ PaimonSnapshotCacheValue fenceValue = fence.getSnapshotCacheValue();
+ PaimonSnapshot fenceSnapshot = fenceValue.getSnapshot();
+ long snapshotId = fenceSnapshot.getSnapshotId();
+ TableScanParams params = scanParams.get();
+ Map<String, String> rawOptions = params.getMapParams();
+ params.reuseResolvedMapParams(PaimonScanParams.pinOptionsToSnapshot(
Review Comment:
[P1] Preserve the fence's schema generation when reusing this synthetic
snapshot pin. `pinOptionsToSnapshot` stores an ordinary `scan.snapshot-id`;
later `getPaimonTable(scanParams)` treats that resolved key as a user schema
selector, switches to `getBasePaimonTable()`, and `applyOptions` runs Paimon's
normal `copy`, which time-travels schema. If S has schema A and a schema-only
ALTER creates B, the fence correctly captures (S, B), but schema binding/scan
decoration reverts this OPTIONS alias to A while a plain alias stays on B. This
is downstream of the loader fix in r3689955956: the corrected fence is already
returned before this copy undoes it. Mark synthetic pins or use
`applyOptionsWithoutTimeTravel` against the fenced table, and test
`getFullSchema` plus split planning after a schema-only ALTER.
##########
fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCacheTest.java:
##########
@@ -68,8 +96,238 @@ public void
testLatestSnapshotUsesLatestSchemaForPinnedRead() {
Assert.assertEquals(12L, value.getSnapshot().getSnapshotId());
Assert.assertEquals(4L, value.getSnapshot().getSchemaId());
Assert.assertSame(pinnedTable, value.getSnapshot().getTable());
-
Mockito.verify(latestSchemaTable).copyWithoutTimeTravel(Collections.singletonMap(
- CoreOptions.SCAN_SNAPSHOT_ID.key(), "12"));
+ Mockito.verify(latestSchemaTable).copyWithoutTimeTravel(
+ Mockito.argThat(options ->
"12".equals(options.get(CoreOptions.SCAN_SNAPSHOT_ID.key()))
+ && options.entrySet().stream()
+ .filter(entry -> entry.getValue() != null)
+ .count() == 1));
+ }
+
+ @Test
+ public void
testFullLatestProjectionCapsManifestParallelismBeforePartitionLoad() throws
AnalysisException {
+ // Keep the test aligned with the loader contract: partition loading
may report analysis failures.
+ int localCapacity = Runtime.getRuntime().availableProcessors();
+ Assume.assumeTrue(localCapacity < 256);
+ PaimonPartitionInfoLoader partitionLoader =
Mockito.mock(PaimonPartitionInfoLoader.class);
+ Mockito.when(partitionLoader.load(Mockito.any(), Mockito.any(),
Mockito.any()))
+ .thenReturn(PaimonPartitionInfo.EMPTY);
+ PaimonLatestSnapshotProjectionLoader loader = new
PaimonLatestSnapshotProjectionLoader(
+ partitionLoader,
+ (nameMapping, schemaId) -> new PaimonSchemaCacheValue(
+ Collections.emptyList(), Collections.emptyList(),
null));
+ NameMapping nameMapping = new NameMapping(1L, "db", "table",
"remote_db", "remote_table");
+ FileStoreTable baseTable = Mockito.mock(FileStoreTable.class);
+ FileStoreTable latestSchemaTable = Mockito.mock(FileStoreTable.class);
+ FileStoreTable pinnedTable = Mockito.mock(FileStoreTable.class);
+ Snapshot snapshot = Mockito.mock(Snapshot.class);
+ SchemaManager schemaManager = Mockito.mock(SchemaManager.class);
+ TableSchema latestSchema = Mockito.mock(TableSchema.class);
+
Mockito.when(baseTable.copyWithLatestSchema()).thenReturn(latestSchemaTable);
+
Mockito.when(latestSchemaTable.options()).thenReturn(Collections.singletonMap(
+ CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), "256"));
+
Mockito.when(latestSchemaTable.latestSnapshot()).thenReturn(Optional.of(snapshot));
+ Mockito.when(snapshot.id()).thenReturn(12L);
+
Mockito.when(latestSchemaTable.copyWithoutTimeTravel(Mockito.anyMap())).thenReturn(pinnedTable);
Review Comment:
[P1] Mock the post-snapshot table that production actually normalizes.
`resolveLatestSnapshot` first gets `pinnedTable =
latestSchemaTable.copyWithoutTimeTravel(snapshotOptions)` and then calls
`runtimeSafeTable(pinnedTable)`. This test puts 256 only on
`latestSchemaTable`; `pinnedTable.options()` defaults empty, so on every host
satisfying `localCapacity < 256`, normalization returns `pinnedTable`
unchanged. The verified cap-bearing call on `latestSchemaTable` never happens
(its only copy carries the snapshot selector), and this test fails. Give
`pinnedTable` the inherited value, stub its cap copy to a separate capped
table, and verify the partition loader receives that table.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/merge/MergeIntoCommand.java:
##########
@@ -121,6 +121,14 @@ public MergeIntoCommand(List<String> targetNameParts,
Optional<String> targetAli
Objects.requireNonNull(notMatchedClauses, "notMatchedClauses
should not be null"));
}
+ /** Return every relation root retained across prepared executions. */
+ public List<LogicalPlan> getRelationRoots() {
Review Comment:
[P1] Expose subquery plans retained by MERGE's own expressions. This method
returns only CTE/source, but `onClause` and matched/not-matched expressions can
contain `SubqueryExpr`, and those query plans are neither plan children nor
expressions under the returned roots. `ExecuteCommand` therefore never clears
an OPTIONS relation in a prepared MERGE ON/WHEN subquery. After the first
EXECUTE pins S and a later commit advances latest, the second EXECUTE resets
`StatementContext` but retains `resolvedMapParams=S`, causing a conflict or
stale selector resolution. This is distinct from r3694727320's source/CTE
omission. Traverse every command-owned expression's subquery plan and test two
EXECUTEs separated by a commit.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonSysExternalTable.java:
##########
@@ -135,23 +153,117 @@ public Table getSysPaimonTable() {
}
public Table getSysPaimonTable(TableScanParams scanParams) {
- Table table = getSysPaimonTable();
+ return getSysPaimonTable(getRawSysPaimonDataTable(), scanParams);
+ }
+
+ public Table getSysPaimonTable(FileStoreTable dataTable, TableScanParams
scanParams) {
if (scanParams == null || !scanParams.isOptions()) {
- return table;
+ FileStoreTable safeDataTable = (FileStoreTable)
PaimonReaderOptions.runtimeSafeTable(dataTable);
Review Comment:
[P2] Apply data-reader validation only to system types that open data
readers. This unconditional check rejects a physical `read.batch-size=0` for
`$schemas`, `$options`, and `$partitions`, although Doris classifies them as
non-Paimon-reader paths and their implementations only read schema/partition
metadata. `$schemas` and `$options` also reject OPTIONS, so a legacy/physical
unsafe reader value cannot be overridden for an otherwise safe metadata query.
The same capability-blind check is repeated in ScanNode/TVF and the old-FE Java
backstop. Keep manifest-planning checks where appropriate, but gate batch/async
reader validation on the system type's reader capability, and cover allowed
`$schemas`/`$partitions` versus reader-backed `$audit_log`.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java:
##########
@@ -519,69 +622,271 @@ static Optional<Long> parseDataSizeBytes(String value) {
if (value == null || value.trim().isEmpty()) {
return Optional.empty();
}
- String normalized = value.trim().toLowerCase(Locale.ROOT).replace("_",
"").replace(" ", "");
- int unitStart = 0;
- while (unitStart < normalized.length()
- && (Character.isDigit(normalized.charAt(unitStart)) ||
normalized.charAt(unitStart) == '.')) {
- unitStart++;
- }
- if (unitStart == 0) {
- return Optional.empty();
- }
try {
- double number = Double.parseDouble(normalized.substring(0,
unitStart));
- String unit = normalized.substring(unitStart);
- long multiplier;
- switch (unit) {
- case "":
- case "b":
- case "byte":
- case "bytes":
- multiplier = 1L;
- break;
- case "k":
- case "kb":
- case "kib":
- multiplier = 1024L;
- break;
- case "m":
- case "mb":
- case "mib":
- multiplier = 1024L * 1024L;
- break;
- case "g":
- case "gb":
- case "gib":
- multiplier = 1024L * 1024L * 1024L;
- break;
- case "t":
- case "tb":
- case "tib":
- multiplier = 1024L * 1024L * 1024L * 1024L;
- break;
- default:
- return Optional.empty();
- }
- return Optional.of((long) (number * multiplier));
- } catch (NumberFormatException e) {
+ // Keep the BE guard's accepted grammar identical to the Paimon
option parser that will
+ // consume this value; accepting a superset lets invalid
serialized options reach scans.
+ return Optional.of(MemorySize.parse(value).getBytes());
+ } catch (IllegalArgumentException e) {
return Optional.empty();
}
}
private void initTable() {
Preconditions.checkState(params.containsKey("serialized_table"));
table = PaimonUtils.deserialize(params.get("serialized_table"));
+ String encodedSystemSource = params.get(PAIMON_OPTION_PREFIX +
DORIS_SERIALIZED_SYSTEM_SOURCE);
+ FileStoreTable systemSource = encodedSystemSource == null
+ ? null : PaimonUtils.deserialize(encodedSystemSource);
+ table = applyBackendManifestParallelism(table,
+ params.get(PAIMON_OPTION_PREFIX +
DORIS_MANIFEST_PARALLELISM_CAP),
+ Runtime.getRuntime().availableProcessors(), systemSource,
+ params.get(PAIMON_OPTION_PREFIX + DORIS_SYSTEM_TABLE_TYPE));
+ table = applyDefaultReadBatchSize(table, batchSize);
+ paimonAllFieldNames = PaimonUtils.getFieldNames(this.table.rowType());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames);
+ }
+ }
+
+ static Table applyDefaultReadBatchSize(Table table, int dorisBatchSize) {
+ validateSerializedReaderOptions(table);
+ if (hasReadBatchSize(table)) {
+ // Doris' output block size and Paimon's reader batch are
independent controls; an
+ // explicitly validated value on any hidden reader must survive
transport unchanged.
+ return table;
+ }
// The serialized table may pin an older data snapshot while carrying
the latest schema
// after a schema change. Applying a normal copy would time travel to
that snapshot's
// schema again and make renamed or newly added columns disappear.
Map<String, String> readOptions = Collections.singletonMap(
- CoreOptions.READ_BATCH_SIZE.key(), String.valueOf(batchSize));
- table = table instanceof FileStoreTable
+ CoreOptions.READ_BATCH_SIZE.key(),
String.valueOf(dorisBatchSize));
+ return table instanceof FileStoreTable
Review Comment:
[P1] Preserve the transported schema generation when injecting the default
batch size into a system wrapper. The preceding rebuild can produce
`AuditLogTable`/`BinlogTable`/`RowTrackingTable`/`ReadOptimizedTable` over a
source fenced at (snapshot S, latest schema B), but this non-FileStore branch
calls the wrapper's normal `copy`. In Paimon 1.3.1 each wrapper delegates to
`wrapped.copy`, whose time-travel path replaces B with S's schema A. With no
explicit batch option, BE initialization can therefore lose a newly
added/renamed field after FE already bound B. This is a later BE-only copy,
distinct from the FE system-wrapper thread r3698463998. Apply the option with
`copyWithoutTimeTravel` to the hidden/transported source and rebuild the
wrapper; test a reader-backed system scan across a schema-only ALTER.
--
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]
