This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 7d16729d1fc [fix](paimon) Preserve schema options across renamed-field
reads (#69545)
7d16729d1fc is described below
commit 7d16729d1fc89dabf0ca898d621baa1d1ea0b30e
Author: Gabriel <[email protected]>
AuthorDate: Sun Oct 25 22:33:45 2026 +0800
[fix](paimon) Preserve schema options across renamed-field reads (#69545)
### What problem does this PR solve?
After renaming a Paimon sequence or aggregation field, current and
historical reads can fail with a missing-field error. Paimon's table
copy selects the historical columns but overlays current table options,
so `sequence.field` and `fields.<column>.aggregate-function` can
reference names that do not exist in the selected schema.
JNI transport also rebuilds tables without their catalog loader. The SDK
factory can reapply a snapshot selector during that rebuild, reverting a
schema-only rename before the bound schema is restored. Build with
selectors temporarily removed and then restore the complete bound
schema.
Restore the selected schema and its physical options together before
applying query options. Preserve the resolved table's read-state keys
and branch policy, including absent keys, so historical options cannot
resurrect a removed scan mode or change fallback priority during
parallelism normalization. Preserve explicit catalog overrides and the
SDK's fallback-branch snapshot translation. Apply the same logic to scan
planning, metadata, statistics, and the source used to rebuild
system-table wrappers, so binding and JNI serialization agree.
Add real-table tests for snapshot, timestamp, tag, and OPTIONS reads
after renames, current reads after schema-only renames, catalog-loader
removal and serialization, catalog override precedence, fallback
branches, empty-table timestamps, and read-optimized system tables. Add
an end-to-end regression covering sequence semantics and repeated
aggregation-field renames with `force_jni_scanner=false` and `true`.
### Release note
Fix Paimon current and historical reads after renaming sequence or
aggregation fields.
### Check List (For Author)
- Test:
- Baseline: 8 historical-read cases fail with missing-field errors. Two
additional catalog-backed current-read cases fail during loader removal.
Review regressions reproduce a removed startup mode being restored and
different branch rows at capacities 64 and 128.
- Full Paimon connector reactor install: 866 tests, 0 failures, 0
errors, 1 existing skipped; all 21 new tests pass.
- FE Checkstyle: passed. FE/BE build and connector packaging: passed.
- End-to-end regression: all 26 query checkpoints passed on a running
FE/BE cluster; generated expected output was reviewed and a second
comparison run passed (1 suite, 0 failures, 0 skips).
- Behavior changed: Yes. Current reads retain their statement-bound
schema during JNI transport, and historical reads use field-dependent
options from the selected schema.
- Does this need documentation: No.
---
.../doris/connector/paimon/PaimonCatalogOps.java | 2 +-
.../connector/paimon/PaimonConnectorMetadata.java | 38 +-
.../connector/paimon/PaimonReaderOptions.java | 10 +-
.../doris/connector/paimon/PaimonScanParams.java | 86 +++-
.../connector/paimon/PaimonScanPlanProvider.java | 31 +-
.../connector/paimon/PaimonTableDecorators.java | 17 +
.../paimon/PaimonHistoricalSchemaOptionsTest.java | 452 +++++++++++++++++++++
.../test_paimon_renamed_field_time_travel.out | 107 +++++
.../test_paimon_renamed_field_time_travel.groovy | 207 ++++++++++
9 files changed, 915 insertions(+), 35 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java
index 48a69bf114f..bd605d69691 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java
@@ -312,7 +312,7 @@ public interface PaimonCatalogOps {
// so this must not invalidate: REFRESH, Doris writes and DDL
invalidate through CatalogMetaCache.
Table table = catalog.getTable(identifier);
Map<String, String> optionsForCopy =
PaimonTableOptions.forCopy(tableOptions);
- return optionsForCopy.isEmpty() ? table :
table.copy(optionsForCopy);
+ return optionsForCopy.isEmpty() ? table :
PaimonScanParams.copyWithHistoricalSchema(table, optionsForCopy);
}
@Override
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
index 79ecda41202..c9de0bba921 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
@@ -420,15 +420,11 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
? paimonHandle
: (PaimonTableHandle) applySnapshot(session, paimonHandle,
snapshot);
Map<String, String> scanOptions = pinned.getScanOptions();
- if (PaimonScanParams.preservesBoundSchema(scanOptions)) {
+ if (PaimonScanParams.preservesBoundSchema(scanOptions) ||
PaimonScanParams.isOptionsPin(scanOptions)) {
// Schema-derived wrappers must be built over the bound source
before exposing their fields.
return new PaimonScanPlanProvider(catalogProperties, catalogOps,
context).resolveScanTable(pinned);
}
- Table table = resolveTable(pinned);
- if (scanOptions != null && !scanOptions.isEmpty() &&
PaimonScanParams.isOptionsPin(scanOptions)) {
- return PaimonScanParams.applyOptions(table, scanOptions);
- }
- return table;
+ return resolveTable(pinned);
}
/**
@@ -870,7 +866,8 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
resolved = PaimonScanParams.pinOptionsToSnapshot(
spec.getOptions(),
spec.getLatestSnapshotFence().getAsLong());
} else {
- resolved = PaimonScanParams.resolveOptions(table,
spec.getOptions());
+ resolved = PaimonScanParams.resolveOptions(table,
spec.getOptions(),
+
PaimonTableOptions.extractCompatible(catalogProperties.getRaw()));
}
String pinnedTag =
resolved.get(CoreOptions.SCAN_TAG_NAME.key());
if (pinnedTag != null) {
@@ -1755,6 +1752,13 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
* the partition columns and escapes path-special characters in the name
via the Paimon SDK.
*/
private List<ConnectorPartitionInfo> collectPartitions(PaimonTableHandle
paimonHandle) {
+ if (paimonHandle.getScanOptions().isEmpty()) {
+ return collectPartitionsAuthenticated(paimonHandle);
+ }
+ return readMetadataAuthenticated(() ->
collectPartitionsAuthenticated(paimonHandle));
+ }
+
+ private List<ConnectorPartitionInfo>
collectPartitionsAuthenticated(PaimonTableHandle paimonHandle) {
if (PaimonScanParams.isPinnedEmptyScan(paimonHandle.getScanOptions()))
{
// Do not reopen latest metadata after the statement fenced an
empty table.
return Collections.emptyList();
@@ -1767,7 +1771,8 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
}
Table resolvedTable = resolveTable(paimonHandle);
- Map<String, String> scanOptions = paimonHandle.getScanOptions();
+ Map<String, String> scanOptions = PaimonScanParams.withCatalogOptions(
+ paimonHandle.getScanOptions(),
PaimonTableOptions.extractCompatible(catalogProperties.getRaw()));
boolean optionsPin = PaimonScanParams.isOptionsPin(scanOptions);
Table table;
if (optionsPin) {
@@ -1778,8 +1783,9 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
// partitions so EXPLAIN and block-rule accounting describe the
same version as the scan.
Table partitionTable = snapshotId == null
? resolvedTable
- : resolvedTable.copy(Collections.singletonMap(
- CoreOptions.SCAN_SNAPSHOT_ID.key(), snapshotId));
+ : PaimonScanParams.copyWithHistoricalSchema(resolvedTable,
PaimonScanParams.withCatalogOptions(
+
Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(), snapshotId),
+
PaimonTableOptions.extractCompatible(catalogProperties.getRaw())));
// Partition projection never opens a data reader, so reader-only
settings must not
// invalidate metadata that a later relation-scoped override can
make safe.
// Metadata planning can also touch Paimon's global manifest
executor, so apply the
@@ -1949,6 +1955,11 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
if (snapshot == null) {
return getTableStatistics(session, handle);
}
+ return readMetadataAuthenticated(() ->
getTableStatisticsAtSnapshot(session, handle, snapshot));
+ }
+
+ private Optional<ConnectorTableStatistics> getTableStatisticsAtSnapshot(
+ ConnectorSession session, ConnectorTableHandle handle,
ConnectorMvccSnapshot snapshot) {
long rowCount;
try {
PaimonTableHandle pinned = (PaimonTableHandle)
applySnapshot(session, handle, snapshot);
@@ -1957,17 +1968,18 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
// first commit even though execution is still required to
scan zero rows.
return Optional.empty();
}
- Map<String, String> scanOptions = pinned.getScanOptions();
+ Map<String, String> scanOptions =
PaimonScanParams.withCatalogOptions(
+ pinned.getScanOptions(),
PaimonTableOptions.extractCompatible(catalogProperties.getRaw()));
// applyOptions removes this Doris-only marker, but execution
still filters files by it.
// This also covers scan.creation-time-millis when it resolves to
a file-creation scan.
if
(PaimonScanParams.getPinnedFileCreationTime(scanOptions).isPresent()) {
return Optional.empty();
}
Table table = resolveTable(pinned);
- if (scanOptions != null && !scanOptions.isEmpty()) {
+ if (scanOptions != null && !scanOptions.isEmpty() &&
!pinned.isSystemTable()) {
table = PaimonScanParams.isOptionsPin(scanOptions)
? PaimonScanParams.applyOptions(table, scanOptions)
- : table.copy(scanOptions);
+ : PaimonScanParams.copyWithHistoricalSchema(table,
scanOptions);
}
table = PaimonReaderOptions.runtimeSafeTable(table);
table = runtimeSafeSystemTable(pinned, table, scanOptions);
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java
index 8907dc38ed0..053eac694db 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java
@@ -189,6 +189,14 @@ public final class PaimonReaderOptions {
}
return rebuilt;
}
+ // Non-file sources cannot be rebuilt through SystemTableLoader;
retain their wrapper's
+ // own copy semantics after independently validating the source's
effective options.
+ if (scanOptions != null && !scanOptions.isEmpty()) {
+ systemTable = PaimonScanParams.isOptionsPin(scanOptions)
+ ? PaimonScanParams.applyOptions(systemTable, scanOptions)
+ : PaimonScanParams.copyWithHistoricalSchema(systemTable,
+
PaimonIncrementalScanParams.applyResetsIfIncremental(scanOptions));
+ }
return runtimeSafeTable(systemTable);
}
@@ -207,7 +215,7 @@ public final class PaimonReaderOptions {
if (scanOptions != null && !scanOptions.isEmpty()) {
// Incremental ranges are relation state too; rebuilding a capped
system wrapper from the
// undecorated source must not silently turn @incr back into a
latest scan.
- effectiveSource = sourceTable.copy(
+ effectiveSource =
PaimonScanParams.copyWithHistoricalSchema(sourceTable,
PaimonIncrementalScanParams.applyResetsIfIncremental(scanOptions));
}
return runtimeSafeTable(effectiveSource);
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java
index 809e371aeac..a05adb64f03 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java
@@ -24,7 +24,9 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.options.ConfigOption;
import org.apache.paimon.options.FallbackKey;
+import org.apache.paimon.options.Options;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.ChainGroupReadTable;
import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
@@ -103,6 +105,13 @@ public final class PaimonScanParams {
private static final Set<String> INHERITED_READ_STATE_KEYS =
buildInheritedReadStateKeys();
+ private static final Set<String> SCHEMA_INDEPENDENT_READ_KEYS =
ImmutableSet.<String>builder()
+ .addAll(INHERITED_READ_STATE_KEYS)
+ .add(CoreOptions.BRANCH.key())
+ .add(CoreOptions.SCAN_PRIMARY_BRANCH.key())
+ .add(CoreOptions.SCAN_FALLBACK_BRANCH.key())
+ .build();
+
private static final Set<String> TIME_TRAVEL_SELECTOR_KEYS =
ImmutableSet.of(
CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
CoreOptions.SCAN_WATERMARK.key(),
@@ -204,13 +213,60 @@ public final class PaimonScanParams {
}
// 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, "");
+ }
+ return dataTable.copy(dynamicOptions);
+ }
+
public static FileStoreTable applyOptionsWithoutTimeTravel(
FileStoreTable table, Map<String, String> options) {
Map<String, String> tableOptions = userOptions(options);
@@ -282,6 +338,17 @@ public final class PaimonScanParams {
merged.put(key, value);
}
});
+ // Read selectors and branch topology belong to the resolved table,
not its historical schema.
+ // Preserve absent keys too: reviving an old mode can invalidate a
snapshot selector, while
+ // reviving an old branch policy can change fallback priority when
reader caps rebuild the pair.
+ SCHEMA_INDEPENDENT_READ_KEYS.forEach(key -> {
+ String value = table.options().get(key);
+ if (value == null) {
+ merged.remove(key);
+ } else {
+ merged.put(key, value);
+ }
+ });
// Explicit catalog overrides can equal old physical values; equality
is not provenance.
options.forEach((key, value) -> {
if (key.startsWith(INTERNAL_PREFIX + "catalog-option.")) {
@@ -353,6 +420,11 @@ public final class PaimonScanParams {
}
public static Map<String, String> resolveOptions(Table table, Map<String,
String> options) {
+ return resolveOptions(table, options,
java.util.Collections.emptyMap());
+ }
+
+ static Map<String, String> resolveOptions(
+ Table table, Map<String, String> options, Map<String, String>
catalogOptions) {
validateOptions(options);
if (!hasStartupOptions(options)) {
return options;
@@ -388,7 +460,7 @@ public final class PaimonScanParams {
return resolveFileCreationTime(options, fileStoreTable,
creationTime);
}
- Table selectedTable = applyOptions(table, options);
+ Table selectedTable = applyOptions(table, withCatalogOptions(options,
catalogOptions));
if
("compacted-full".equalsIgnoreCase(options.get(CoreOptions.SCAN_MODE.key()))) {
Long snapshotId = compactedFullSnapshotId((FileStoreTable)
selectedTable);
return snapshotId == null
@@ -508,9 +580,13 @@ public final class PaimonScanParams {
}
private static Map<String, String> userOptions(Map<String, String>
options) {
- return options.entrySet().stream()
- .filter(entry -> !entry.getKey().startsWith(INTERNAL_PREFIX))
- .collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue));
+ Map<String, String> result = new HashMap<>();
+ options.forEach((key, value) -> {
+ if (!key.startsWith(INTERNAL_PREFIX)) {
+ result.put(key, value);
+ }
+ });
+ return result;
}
/**
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
index 6c5493765e1..548907b5249 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
@@ -361,10 +361,11 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
}
private <T> T withBoundSchemaAuthentication(PaimonTableHandle handle,
Supplier<T> action) {
- if (context == null ||
!PaimonScanParams.preservesBoundSchema(handle.getScanOptions())) {
+ if (context == null || (handle.getScanOptions().isEmpty()
+ &&
PaimonTableOptions.extractCompatible(catalogProps.getRaw()).isEmpty())) {
return action.get();
}
- // Restoring a bound schema can read FileIO after table resolution has
left the authenticated scope.
+ // Historical and statement schema restoration read FileIO after table
resolution leaves its scope.
try {
return context.executeAuthenticated(action::get);
} catch (RuntimeException e) {
@@ -376,20 +377,17 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
private Map<String, String> effectiveScanOptions(PaimonTableHandle handle)
{
Map<String, String> options = handle.getScanOptions();
- if (PaimonScanParams.preservesBoundSchema(options)) {
- // Replayed catalogs already ignored legacy options during
binding; scans must use the same policy.
- return PaimonScanParams.withCatalogOptions(options,
-
PaimonTableOptions.extractCompatible(catalogProps.getRaw()));
- }
- return options;
+ // An explicit catalog value may equal latest's value but differ from
the historical schema.
+ return PaimonScanParams.withCatalogOptions(options,
+ PaimonTableOptions.extractCompatible(catalogProps.getRaw()));
}
private Table applyScanOptions(PaimonTableHandle paimonHandle, Table
table) {
Map<String, String> scanOptions = effectiveScanOptions(paimonHandle);
Table finalTable = table;
- if (scanOptions != null && !scanOptions.isEmpty()
- && !(paimonHandle.isSystemTable() &&
PaimonScanParams.preservesBoundSchema(scanOptions))) {
- // Statement-fenced system wrappers are rebuilt below from their
schema-bound source.
+ if (scanOptions != null && !scanOptions.isEmpty() &&
!paimonHandle.isSystemTable()) {
+ // System wrappers are rebuilt below from their schema-bound
source. Copying the wrapper
+ // first would time-travel inside the SDK before historical field
options can be restored.
if (table instanceof FileStoreTable
&& PaimonScanParams.preservesBoundSchema(scanOptions)) {
// A statement fence owns data visibility, not schema time
travel. Reusing Table.copy
@@ -406,7 +404,8 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
// scan.snapshot-id/scan.mode here (the single Table.copy
chokepoint shared by both the
// native/JNI scan path and the JNI serialized-table path) so
a stale persisted pin on the
// base table cannot hijack incremental-between.
Non-incremental pins pass through unchanged.
- finalTable =
table.copy(PaimonIncrementalScanParams.applyResetsIfIncremental(scanOptions));
+ finalTable = PaimonScanParams.copyWithHistoricalSchema(table,
+
PaimonIncrementalScanParams.applyResetsIfIncremental(scanOptions));
}
}
finalTable = PaimonReaderOptions.runtimeSafeTable(finalTable);
@@ -1799,10 +1798,12 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
}
private static FileStoreTable rebuildWithoutCatalogLoader(FileStoreTable
branch) {
- // The factory evaluates scan.snapshot-id and can rewind a schema-only
ALTER. Restore
- // the already-bound schema after removing the loader so JNI reads the
planned field ids.
+ // The factory invokes copy(Map) before returning, so selectors can
rewind a schema-only
+ // ALTER and fail field-option validation before the bound schema can
be restored.
+ Map<String, String> factoryOptions = new
HashMap<>(branch.schema().options());
+
PaimonScanParams.inheritedReadStateKeys().forEach(factoryOptions::remove);
return FileStoreTableFactory.createWithoutFallbackBranch(
- branch.fileIO(), branch.location(), branch.schema(), new
Options(),
+ branch.fileIO(), branch.location(),
branch.schema().copy(factoryOptions), new Options(),
CatalogEnvironment.empty()).copy(branch.schema());
}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java
index 48e2e11ad94..f6f17e34b3c 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java
@@ -23,6 +23,8 @@ import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
+import java.util.Map;
+
/**
* The one place that knows how Paimon stacks {@link DelegatedFileStoreTable}
decorators on a loaded
* table, and how far they may be peeled off.
@@ -45,6 +47,21 @@ final class PaimonTableDecorators {
return new FallbackReadFileStoreTable(wrapped, other,
PaimonReaderOptions.isWrappedFirst(pair));
}
+ static Map<String, String> otherOptions(FallbackReadFileStoreTable pair,
Map<String, String> options) {
+ return new FallbackOptions(pair).rewriteOtherOptions(options);
+ }
+
+ private static final class FallbackOptions extends
FallbackReadFileStoreTable {
+ private FallbackOptions(FallbackReadFileStoreTable pair) {
+ super(pair.wrapped(), pair.other(),
PaimonReaderOptions.isWrappedFirst(pair));
+ }
+
+ @Override
+ public Map<String, String> rewriteOtherOptions(Map<String, String>
options) {
+ return super.rewriteOtherOptions(options);
+ }
+ }
+
/**
* Whether {@code table} reads as a chain table: a fallback pair whose
other branch is a
* {@link ChainGroupReadTable}. A branch of a chain table read on its own
is a plain table.
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonHistoricalSchemaOptionsTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonHistoricalSchemaOptionsTest.java
new file mode 100644
index 00000000000..3ab8f855e38
--- /dev/null
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonHistoricalSchemaOptionsTest.java
@@ -0,0 +1,452 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.connector.paimon;
+
+import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot;
+import org.apache.doris.connector.spi.mvcc.ConnectorTimeTravelSpec;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.FileSystemCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.CatalogEnvironment;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.system.SystemTableLoader;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.InstantiationUtil;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class PaimonHistoricalSchemaOptionsTest {
+ @Test
+ public void snapshotReadDoesNotRestoreRemovedStartupMode(@TempDir Path
warehouse) throws Exception {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "removed_startup_mode");
+ catalog.createTable(id, Schema.newBuilder().column("id",
DataTypes.INT().notNull())
+ .column("value", DataTypes.INT()).primaryKey("id")
+ .option("bucket", "1").option("file.format", "parquet")
+ .option("scan.mode", "latest-full").build(), false);
+ append((FileStoreTable) catalog.getTable(id), GenericRow.of(1,
10));
+ catalog.alterTable(id,
Collections.singletonList(SchemaChange.removeOption("scan.mode")), false);
+ Table latest = catalog.getTable(id);
+ Assertions.assertFalse(latest.options().containsKey("scan.mode"));
+ Map<String, String> selector =
Collections.singletonMap("scan.snapshot-id", "1");
+ // Removing a startup mode must remain effective when a historical
schema is restored.
+ Assertions.assertEquals(Collections.singletonList("1:10"),
readRows(latest.copy(selector)));
+ Table historical =
PaimonScanParams.copyWithHistoricalSchema(latest, selector);
+ Assertions.assertEquals(Collections.singletonList("1:10"),
readRows(historical));
+ }
+ }
+
+ @Test
+ public void historicalFallbackPriorityIsStableAt64(@TempDir Path
warehouse) throws Exception {
+ checkHistoricalFallbackPriority(warehouse, 64);
+ }
+
+ @Test
+ public void historicalFallbackPriorityIsStableAt128(@TempDir Path
warehouse) throws Exception {
+ checkHistoricalFallbackPriority(warehouse, 128);
+ }
+
+ private void checkHistoricalFallbackPriority(Path warehouse, int capacity)
throws Exception {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "branch_priority_history");
+ catalog.createTable(id, Schema.newBuilder().column("id",
DataTypes.INT().notNull())
+ .column("value",
DataTypes.INT().notNull()).primaryKey("id", "value").partitionKeys("id")
+ .option("bucket", "1").option("file.format",
"parquet").build(), false);
+ FileStoreTable original = (FileStoreTable) catalog.getTable(id);
+ original.createBranch("backup");
+ Identifier backupId = new Identifier("db",
"branch_priority_history", "backup");
+ append((FileStoreTable) catalog.getTable(backupId),
GenericRow.of(1, 20));
+ catalog.alterTable(id,
Arrays.asList(SchemaChange.setOption("scan.primary-branch", "backup"),
+ SchemaChange.setOption("scan.manifest.parallelism",
"128")), false);
+ FallbackReadFileStoreTable primary = (FallbackReadFileStoreTable)
catalog.getTable(id);
+ append(primary.wrapped(), GenericRow.of(1, 10));
+ Assertions.assertEquals(Collections.singletonList("1:20"),
readRows(primary));
+ catalog.alterTable(id,
Arrays.asList(SchemaChange.removeOption("scan.primary-branch"),
+ SchemaChange.setOption("scan.fallback-branch", "backup")),
false);
+ FallbackReadFileStoreTable latest = (FallbackReadFileStoreTable)
catalog.getTable(id);
+ Assertions.assertEquals(Collections.singletonList("1:10"),
readRows(latest));
+ Table historical =
PaimonScanParams.copyWithHistoricalSchema(latest,
+ Collections.singletonMap("scan.snapshot-id", "1"));
+ Assertions.assertEquals("backup",
historical.options().get("scan.fallback-branch"));
+
Assertions.assertFalse(historical.options().containsKey("scan.primary-branch"));
+ // The same overlapping partition must choose the same branch on
different FE capacities.
+ Table normalized =
PaimonReaderOptions.runtimeSafeTable(historical, capacity);
+ Assertions.assertEquals(Integer.toString(capacity),
normalized.options().get("scan.manifest.parallelism"));
+ Assertions.assertEquals("backup",
normalized.options().get("scan.fallback-branch"));
+
Assertions.assertFalse(normalized.options().containsKey("scan.primary-branch"));
+ Assertions.assertEquals(Collections.singletonList("1:10"),
readRows(normalized));
+ }
+ }
+
+ @Test
+ public void snapshotUsesHistoricalSequenceFieldAfterRename(@TempDir Path
warehouse) throws Exception {
+ checkRenamedField(warehouse, false, "snapshot");
+ }
+
+ @Test
+ public void timestampUsesHistoricalSequenceFieldAfterRename(@TempDir Path
warehouse) throws Exception {
+ checkRenamedField(warehouse, false, "timestamp");
+ }
+
+ @Test
+ public void tagUsesHistoricalSequenceFieldAfterRename(@TempDir Path
warehouse) throws Exception {
+ checkRenamedField(warehouse, false, "tag");
+ }
+
+ @Test
+ public void optionsUsesHistoricalSequenceFieldAfterRename(@TempDir Path
warehouse) throws Exception {
+ checkRenamedField(warehouse, false, "options");
+ }
+
+ @Test
+ public void currentReadUsesRenamedSequenceFieldFromWarmTable(@TempDir Path
warehouse) throws Exception {
+ checkRenamedField(warehouse, false, "current");
+ }
+
+ @Test
+ public void snapshotUsesHistoricalAggregationFieldAfterRename(@TempDir
Path warehouse) throws Exception {
+ checkRenamedField(warehouse, true, "snapshot");
+ }
+
+ @Test
+ public void timestampUsesHistoricalAggregationFieldAfterRename(@TempDir
Path warehouse) throws Exception {
+ checkRenamedField(warehouse, true, "timestamp");
+ }
+
+ @Test
+ public void tagUsesHistoricalAggregationFieldAfterRename(@TempDir Path
warehouse) throws Exception {
+ checkRenamedField(warehouse, true, "tag");
+ }
+
+ @Test
+ public void optionsUsesHistoricalAggregationFieldAfterRename(@TempDir Path
warehouse) throws Exception {
+ checkRenamedField(warehouse, true, "options");
+ }
+
+ @Test
+ public void currentReadUsesRenamedAggregationFieldFromWarmTable(@TempDir
Path warehouse) throws Exception {
+ checkRenamedField(warehouse, true, "current");
+ }
+
+ @Test
+ public void systemOptionsUsesHistoricalSequenceFieldAfterRename(@TempDir
Path warehouse) throws Exception {
+ checkRenamedField(warehouse, false, "system");
+ }
+
+ @Test
+ public void historicalSchemaKeepsExplicitCatalogReaderOption(@TempDir Path
warehouse) throws Exception {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "reader_history");
+ catalog.createTable(id, Schema.newBuilder().column("id",
DataTypes.INT().notNull())
+ .column("value",
DataTypes.INT().notNull()).primaryKey("id", "value").partitionKeys("id")
+ .option("bucket", "1").option("sequence.field", "value")
+ .option("read.batch-size", "64").option("file.format",
"parquet").build(), false);
+ FileStoreTable original = (FileStoreTable) catalog.getTable(id);
+ append(original, GenericRow.of(1, 10));
+ catalog.alterTable(id,
Arrays.asList(SchemaChange.renameColumn("value", "renamed_value"),
+ SchemaChange.setOption("read.batch-size", "32")), false);
+ Map<String, String> properties =
Collections.singletonMap("paimon.table-option.read.batch-size", "32");
+ PaimonCatalogOps ops = new
PaimonCatalogOps.CatalogBackedPaimonCatalogOps(
+ catalog, Collections.singletonMap("read.batch-size",
"32"));
+ PaimonCatalogProperties catalogProperties =
PaimonCatalogProperties.of(properties);
+ PaimonConnectorMetadata metadata = new PaimonConnectorMetadata(
+ ops, catalogProperties, new RecordingConnectorContext());
+ PaimonTableHandle handle = (PaimonTableHandle)
metadata.getTableHandle(null, "db", "reader_history").get();
+ ConnectorMvccSnapshot snapshot = metadata.resolveTimeTravel(null,
handle,
+ ConnectorTimeTravelSpec.snapshotId("1")).get();
+ PaimonTableHandle pinned = (PaimonTableHandle)
metadata.applySnapshot(null, handle, snapshot);
+ Assertions.assertEquals(Collections.singletonList("id=1"),
metadata.listPartitionNames(null, pinned));
+ Assertions.assertTrue(metadata.getTableStatistics(null, handle,
snapshot).isPresent());
+ Table scan = new PaimonScanPlanProvider(catalogProperties,
ops).resolveScanTable(pinned);
+ // Equal current/catalog values still carry an explicit override
of the historical default.
+ Assertions.assertEquals("32",
scan.options().get("read.batch-size"));
+ Assertions.assertEquals(Collections.singletonList("1:10"),
readRows(scan));
+ }
+ }
+
+ @Test
+ public void fallbackBranchesRestoreIndependentHistoricalOptions(@TempDir
Path warehouse) throws Exception {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "fallback_history");
+ catalog.createTable(id, Schema.newBuilder().column("id",
DataTypes.INT().notNull())
+ .column("value", DataTypes.INT()).primaryKey("id")
+ .option("bucket", "1").option("sequence.field", "value")
+ .option("file.format", "parquet").build(), false);
+ FileStoreTable original = (FileStoreTable) catalog.getTable(id);
+ original.createBranch("backup");
+ Identifier backupId = new Identifier("db", "fallback_history",
"backup");
+ FileStoreTable backup = (FileStoreTable)
catalog.getTable(backupId);
+ append(backup, GenericRow.of(2, 8));
+ append(original, GenericRow.of(1, 10));
+ catalog.alterTable(backupId, Collections.singletonList(
+ SchemaChange.renameColumn("value", "renamed_value")),
false);
+ catalog.alterTable(id, Collections.singletonList(
+ SchemaChange.renameColumn("value", "renamed_value")),
false);
+ FileStoreTable latest = (FileStoreTable) catalog.getTable(id);
+ FileStoreTable latestBackup = (FileStoreTable)
catalog.getTable(backupId);
+ append(latestBackup, GenericRow.of(2, 18));
+ append(latest, GenericRow.of(1, 20));
+ FallbackReadFileStoreTable pair = new
FallbackReadFileStoreTable(latest, latestBackup, false);
+ Map<String, String> options = new HashMap<>();
+ options.put("scan.snapshot-id", "1");
+ options.put("branch", "main");
+ FallbackReadFileStoreTable historical =
(FallbackReadFileStoreTable)
+ PaimonScanParams.copyWithHistoricalSchema(pair, options);
+ Assertions.assertEquals("value",
historical.wrapped().options().get("sequence.field"));
+ Assertions.assertEquals("value",
historical.other().options().get("sequence.field"));
+ Assertions.assertEquals("backup",
historical.other().coreOptions().branch());
+ Assertions.assertSame(latest.catalogEnvironment(),
historical.wrapped().catalogEnvironment());
+ Assertions.assertSame(latestBackup.catalogEnvironment(),
historical.other().catalogEnvironment());
+ Assertions.assertEquals(Collections.singletonList("1:10"),
readRows(historical.wrapped()));
+ Assertions.assertEquals(Collections.singletonList("2:8"),
readRows(historical.other()));
+ }
+ }
+
+ @Test
+ public void timestampOnEmptyTableKeepsSdkBehavior(@TempDir Path warehouse)
throws Exception {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "empty_history");
+ catalog.createTable(id, Schema.newBuilder().column("id",
DataTypes.INT()).build(), false);
+ Table table = catalog.getTable(id);
+ Map<String, String> options =
Collections.singletonMap("scan.timestamp-millis", "1");
+ Table selected = PaimonScanParams.copyWithHistoricalSchema(table,
options);
+ Assertions.assertEquals(table.copy(options).rowType(),
selected.rowType());
+ IllegalArgumentException expected = Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
readRows(table.copy(options)));
+ IllegalArgumentException actual = Assertions.assertThrows(
+ IllegalArgumentException.class, () -> readRows(selected));
+ Assertions.assertEquals(expected.getMessage(),
actual.getMessage());
+ }
+ }
+
+ @Test
+ public void
catalogBackedSchemaOnlyRenameKeepsStatementSchemaOnBackend(@TempDir Path
warehouse)
+ throws Exception {
+ checkCatalogBackedTransport(warehouse, true, false);
+ }
+
+ @Test
+ public void
catalogBackedHistoricalReadKeepsHistoricalSchemaOnBackend(@TempDir Path
warehouse)
+ throws Exception {
+ checkCatalogBackedTransport(warehouse, false, false);
+ }
+
+ @Test
+ public void
catalogBackedAggregationRenameKeepsStatementSchemaOnBackend(@TempDir Path
warehouse)
+ throws Exception {
+ checkCatalogBackedTransport(warehouse, true, true);
+ }
+
+ @Test
+ public void
catalogBackedAggregationHistoryKeepsHistoricalSchemaOnBackend(@TempDir Path
warehouse)
+ throws Exception {
+ checkCatalogBackedTransport(warehouse, false, true);
+ }
+
+ private void checkCatalogBackedTransport(Path warehouse, boolean current,
boolean aggregate) throws Exception {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "transport_history");
+ Schema.Builder schema = Schema.newBuilder().column("id",
DataTypes.INT().notNull())
+ .column("value", DataTypes.INT()).primaryKey("id")
+ .option("bucket", "1").option("file.format", "parquet");
+ if (aggregate) {
+ schema.option("merge-engine",
"aggregation").option("fields.value.aggregate-function", "sum");
+ } else {
+ schema.option("sequence.field", "value");
+ }
+ catalog.createTable(id, schema.build(), false);
+ FileStoreTable original = (FileStoreTable) catalog.getTable(id);
+ append(original, GenericRow.of(1, 10));
+ catalog.alterTable(id, Collections.singletonList(
+ SchemaChange.renameColumn("value", "renamed_value")),
false);
+ FileStoreTable latest = (FileStoreTable) catalog.getTable(id);
+ CatalogEnvironment environment = new CatalogEnvironment(id, null,
() -> catalog,
+ null, null, null, false, false);
+ latest =
FileStoreTableFactory.createWithoutFallbackBranch(latest.fileIO(),
latest.location(),
+ latest.schema(), new Options(), environment);
+ PaimonCatalogOps ops = new
PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog);
+ PaimonCatalogProperties properties =
PaimonCatalogProperties.of(Collections.emptyMap());
+ PaimonConnectorMetadata metadata = new PaimonConnectorMetadata(
+ ops, properties, new RecordingConnectorContext());
+ PaimonTableHandle handle = new PaimonTableHandle("db",
"transport_history",
+ Collections.emptyList(), Collections.emptyList());
+ handle.setPaimonTable(latest);
+ ConnectorMvccSnapshot snapshot = current ?
metadata.beginQuerySnapshot(null, handle).get()
+ : metadata.resolveTimeTravel(null, handle,
ConnectorTimeTravelSpec.snapshotId("1")).get();
+ PaimonTableHandle pinned = (PaimonTableHandle)
metadata.applySnapshot(null, handle, snapshot);
+ PaimonScanPlanProvider provider = new
PaimonScanPlanProvider(properties, ops);
+ FileStoreTable scan = (FileStoreTable)
provider.resolveScanTable(pinned);
+ String expectedName = current ? "renamed_value" : "value";
+
Assertions.assertNotNull(scan.catalogEnvironment().catalogLoader());
+ Assertions.assertEquals(Arrays.asList("id", expectedName),
scan.rowType().getFieldNames());
+ // A filesystem-only fixture skips loader removal; serialize the
catalog-backed path too.
+ FileStoreTable backend = InstantiationUtil.deserializeObject(
+
InstantiationUtil.serializeObject(provider.tableForBackend(pinned, scan)),
+ getClass().getClassLoader());
+
Assertions.assertNull(backend.catalogEnvironment().catalogLoader());
+ Assertions.assertEquals(scan.schema(), backend.schema());
+ Assertions.assertEquals("1",
backend.options().get("scan.snapshot-id"));
+ if (aggregate) {
+ Assertions.assertEquals("sum", backend.options().get("fields."
+ expectedName + ".aggregate-function"));
+ } else {
+ Assertions.assertEquals(expectedName,
backend.options().get("sequence.field"));
+ }
+ Assertions.assertEquals(Collections.singletonList("1:10"),
readRows(backend));
+ }
+ }
+
+ private void checkRenamedField(Path warehouse, boolean aggregate, String
selector) throws Exception {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "field_history");
+ Schema.Builder schema = Schema.newBuilder().column("id",
DataTypes.INT().notNull())
+ .column("value", DataTypes.INT()).primaryKey("id")
+ .option("bucket", "1").option("file.format", "parquet");
+ if (aggregate) {
+ schema.option("merge-engine", "aggregation")
+ .option("fields.value.aggregate-function", "sum");
+ } else {
+ schema.option("sequence.field", "value");
+ }
+ catalog.createTable(id, schema.build(), false);
+ FileStoreTable warm = (FileStoreTable) catalog.getTable(id);
+ append(warm, GenericRow.of(1, 10));
+ // A second value proves that the configured merge rule remains
effective during reads.
+ append(warm, GenericRow.of(1, 5));
+ long snapshotId = warm.snapshotManager().latestSnapshotId();
+ long timestamp =
warm.snapshotManager().snapshot(snapshotId).timeMillis();
+ warm.createTag("before_rename", snapshotId);
+ // Read-optimized views can omit un-compacted files; preserve
their real SDK baseline.
+ List<String> systemBaseline = "system".equals(selector)
+ ? readRows(SystemTableLoader.load("ro", warm)) :
Collections.emptyList();
+ catalog.alterTable(id, Collections.singletonList(
+ SchemaChange.renameColumn("value", "renamed_value")),
false);
+ FileStoreTable latest = (FileStoreTable) catalog.getTable(id);
+ append(latest, GenericRow.of(1, 20));
+ String historicalRow = aggregate ? "1:15" : "1:10";
+ String latestRow = aggregate ? "1:35" : "1:20";
+ Assertions.assertEquals(Collections.singletonList(latestRow),
readRows(latest));
+
+ PaimonCatalogOps ops = new
PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog);
+ PaimonCatalogProperties properties =
PaimonCatalogProperties.of(Collections.emptyMap());
+ PaimonConnectorMetadata metadata = new PaimonConnectorMetadata(
+ ops, properties, new RecordingConnectorContext());
+ PaimonTableHandle handle = new PaimonTableHandle("db",
"field_history",
+ Collections.emptyList(), Collections.emptyList());
+ boolean current = "current".equals(selector);
+ handle.setPaimonTable(current ? warm : latest);
+ // Current reads rebind a warm table; explicit selectors must
restore historical field options.
+ ConnectorMvccSnapshot snapshot = current
+ ? metadata.beginQuerySnapshot(null, handle).get()
+ : metadata.resolveTimeTravel(null, handle,
selector(selector, snapshotId, timestamp)).get();
+ if ("system".equals(selector)) {
+ handle = (PaimonTableHandle) metadata.getSysTableHandle(null,
handle, "ro").get();
+ }
+ Assertions.assertEquals(current ? "renamed_value" : "value",
+ metadata.getTableSchema(null, handle,
snapshot).getColumns().get(1).getName());
+ PaimonTableHandle pinned = (PaimonTableHandle)
metadata.applySnapshot(null, handle, snapshot);
+ PaimonScanPlanProvider provider = new
PaimonScanPlanProvider(properties, ops);
+ Table scan = provider.resolveScanTable(pinned);
+ String expectedName = current ? "renamed_value" : "value";
+ String expectedRow = current ? latestRow : historicalRow;
+ Assertions.assertEquals(Arrays.asList("id", expectedName),
scan.rowType().getFieldNames());
+ if (aggregate) {
+ Assertions.assertEquals("sum", scan.options().get("fields." +
expectedName + ".aggregate-function"));
+ } else {
+ Assertions.assertEquals(expectedName,
scan.options().get("sequence.field"));
+ }
+ List<String> expectedRows = "system".equals(selector)
+ ? systemBaseline : Collections.singletonList(expectedRow);
+ Assertions.assertEquals(expectedRows, readRows(scan));
+ Assertions.assertEquals(expectedRows,
readRows(provider.tableForBackend(pinned, scan)));
+ }
+ }
+
+ private static ConnectorTimeTravelSpec selector(String selector, long
snapshotId, long timestamp) {
+ switch (selector) {
+ case "snapshot":
+ return
ConnectorTimeTravelSpec.snapshotId(Long.toString(snapshotId));
+ case "timestamp":
+ return
ConnectorTimeTravelSpec.timestamp(Long.toString(timestamp), true);
+ case "tag":
+ return ConnectorTimeTravelSpec.tag("before_rename");
+ case "system":
+ case "options":
+ return ConnectorTimeTravelSpec.options(
+ Collections.singletonMap("scan.snapshot-id",
Long.toString(snapshotId)));
+ default:
+ throw new IllegalArgumentException("Unexpected test selector:
" + selector);
+ }
+ }
+
+ private static void append(FileStoreTable table, GenericRow row) throws
Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite(); BatchTableCommit
commit = builder.newCommit()) {
+ write.write(row);
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private static List<String> readRows(Table table) throws Exception {
+ List<String> rows = new ArrayList<>();
+ for (Split split : table.newReadBuilder().newScan().plan().splits()) {
+ try (RecordReader<InternalRow> reader =
table.newReadBuilder().newRead().createReader(split)) {
+ reader.forEachRemaining(row -> rows.add(row.getInt(0) + ":" +
row.getInt(1)));
+ }
+ }
+ Collections.sort(rows);
+ return rows;
+ }
+}
diff --git
a/regression-test/data/external_table_p0/paimon/test_paimon_renamed_field_time_travel.out
b/regression-test/data/external_table_p0/paimon/test_paimon_renamed_field_time_travel.out
new file mode 100644
index 00000000000..881f364cfa0
--- /dev/null
+++
b/regression-test/data/external_table_p0/paimon/test_paimon_renamed_field_time_travel.out
@@ -0,0 +1,107 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !sequence_schema_only_false --
+1 10 base
+2 20 retained
+
+-- !sequence_schema_only_true --
+1 10 base
+2 20 retained
+
+-- !aggregation_schema_only_false --
+1 10
+2 20
+
+-- !aggregation_schema_only_true --
+1 10
+2 20
+
+-- !sequence_latest_false --
+1 30 updated
+2 20 retained
+3 15 new
+
+-- !sequence_old_snapshot_false --
+1 10 base
+2 20 retained
+
+-- !sequence_new_snapshot_false --
+1 30 updated
+2 20 retained
+3 15 new
+
+-- !sequence_options_false --
+1 10 base
+2 20 retained
+
+-- !sequence_tag_false --
+1 10 base
+2 20 retained
+
+-- !aggregation_latest_false --
+1 22
+2 27
+
+-- !aggregation_old_snapshot_false --
+1 10
+2 20
+
+-- !aggregation_intermediate_snapshot_false --
+1 15
+2 23
+
+-- !aggregation_options_false --
+1 10
+2 20
+
+-- !aggregation_tag_false --
+1 10
+2 20
+
+-- !removed_scan_mode_false --
+1 10
+
+-- !sequence_latest_true --
+1 30 updated
+2 20 retained
+3 15 new
+
+-- !sequence_old_snapshot_true --
+1 10 base
+2 20 retained
+
+-- !sequence_new_snapshot_true --
+1 30 updated
+2 20 retained
+3 15 new
+
+-- !sequence_options_true --
+1 10 base
+2 20 retained
+
+-- !sequence_tag_true --
+1 10 base
+2 20 retained
+
+-- !aggregation_latest_true --
+1 22
+2 27
+
+-- !aggregation_old_snapshot_true --
+1 10
+2 20
+
+-- !aggregation_intermediate_snapshot_true --
+1 15
+2 23
+
+-- !aggregation_options_true --
+1 10
+2 20
+
+-- !aggregation_tag_true --
+1 10
+2 20
+
+-- !removed_scan_mode_true --
+1 10
+
diff --git
a/regression-test/suites/external_table_p0/paimon/test_paimon_renamed_field_time_travel.groovy
b/regression-test/suites/external_table_p0/paimon/test_paimon_renamed_field_time_travel.groovy
new file mode 100644
index 00000000000..5151cc45934
--- /dev/null
+++
b/regression-test/suites/external_table_p0/paimon/test_paimon_renamed_field_time_travel.groovy
@@ -0,0 +1,207 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_paimon_renamed_field_time_travel", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test")
+ return
+ }
+
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String catalogName = "test_paimon_renamed_field_time_travel"
+ String dbName = "paimon_renamed_field_time_travel_db"
+
+ def latestSnapshotId = { String tableName ->
+ def rows = spark_paimon """
+ SELECT snapshot_id FROM paimon.${dbName}.`${tableName}\$snapshots`
+ ORDER BY snapshot_id DESC LIMIT 1
+ """
+ assertEquals(1, rows.size())
+ return rows[0][0].toString()
+ }
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type'='paimon',
+ 'warehouse'='s3://warehouse/wh',
+ 's3.endpoint'='http://${externalEnvIp}:${minioPort}',
+ 's3.access_key'='admin',
+ 's3.secret_key'='password',
+ 's3.path.style.access'='true'
+ )
+ """
+
+ try {
+ spark_paimon_multi """
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+ DROP TABLE IF EXISTS paimon.${dbName}.sequence_timeline;
+ CREATE TABLE paimon.${dbName}.sequence_timeline (
+ id INT, old_sequence BIGINT, payload STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key'='id',
+ 'bucket'='1',
+ 'file.format'='parquet',
+ 'sequence.field'='old_sequence'
+ );
+ INSERT INTO paimon.${dbName}.sequence_timeline
+ VALUES (1, 10, 'base'), (2, 20, 'retained');
+ """
+ String sequenceBase = latestSnapshotId("sequence_timeline")
+ spark_paimon_multi """
+ CALL paimon.sys.create_tag(
+ table => '${dbName}.sequence_timeline', tag => 'sequence_base'
+ );
+ ALTER TABLE paimon.${dbName}.sequence_timeline
+ RENAME COLUMN old_sequence TO current_sequence;
+ """
+
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+ sql """refresh table sequence_timeline"""
+ // A schema-only rename updates sequence.field without creating a data
snapshot.
+ // Latest reads must keep those two pieces of metadata from the same
schema.
+ assertEquals(sequenceBase, latestSnapshotId("sequence_timeline"))
+ [false, true].each { forceJni ->
+ sql """set force_jni_scanner=${forceJni}"""
+ "order_qt_sequence_schema_only_${forceJni}" """
+ SELECT id, current_sequence, payload FROM sequence_timeline
ORDER BY id
+ """
+ }
+
+ spark_paimon """
+ INSERT INTO paimon.${dbName}.sequence_timeline VALUES
+ (1, 30, 'updated'), (2, 5, 'stale'), (3, 15, 'new')
+ """
+ String sequenceCurrent = latestSnapshotId("sequence_timeline")
+ sql """refresh table sequence_timeline"""
+
+ spark_paimon_multi """
+ DROP TABLE IF EXISTS paimon.${dbName}.aggregation_timeline;
+ CREATE TABLE paimon.${dbName}.aggregation_timeline (
+ id INT, original_total BIGINT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key'='id',
+ 'bucket'='1',
+ 'file.format'='parquet',
+ 'merge-engine'='aggregation',
+ 'fields.original_total.aggregate-function'='sum'
+ );
+ INSERT INTO paimon.${dbName}.aggregation_timeline VALUES (1, 10),
(2, 20);
+ """
+ String aggregationBase = latestSnapshotId("aggregation_timeline")
+ spark_paimon_multi """
+ CALL paimon.sys.create_tag(
+ table => '${dbName}.aggregation_timeline', tag =>
'aggregation_base'
+ );
+ ALTER TABLE paimon.${dbName}.aggregation_timeline
+ RENAME COLUMN original_total TO intermediate_total;
+ """
+ sql """refresh catalog ${catalogName}"""
+ assertEquals(aggregationBase, latestSnapshotId("aggregation_timeline"))
+ [false, true].each { forceJni ->
+ sql """set force_jni_scanner=${forceJni}"""
+ "order_qt_aggregation_schema_only_${forceJni}" """
+ SELECT id, intermediate_total FROM aggregation_timeline ORDER
BY id
+ """
+ }
+ spark_paimon """
+ INSERT INTO paimon.${dbName}.aggregation_timeline VALUES (1, 5),
(2, 3)
+ """
+ String aggregationIntermediate =
latestSnapshotId("aggregation_timeline")
+ spark_paimon_multi """
+ ALTER TABLE paimon.${dbName}.aggregation_timeline
+ RENAME COLUMN intermediate_total TO current_total;
+ INSERT INTO paimon.${dbName}.aggregation_timeline VALUES (1, 7),
(2, 4);
+ """
+ sql """refresh catalog ${catalogName}"""
+
+ spark_paimon_multi """
+ DROP TABLE IF EXISTS paimon.${dbName}.scan_state_history;
+ CREATE TABLE paimon.${dbName}.scan_state_history (
+ id INT, value BIGINT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key'='id', 'bucket'='1', 'file.format'='parquet',
+ 'scan.mode'='latest-full'
+ );
+ INSERT INTO paimon.${dbName}.scan_state_history VALUES (1, 10);
+ """
+ String scanStateSnapshot = latestSnapshotId("scan_state_history")
+ spark_paimon """
+ ALTER TABLE paimon.${dbName}.scan_state_history UNSET
TBLPROPERTIES ('scan.mode')
+ """
+ sql """refresh catalog ${catalogName}"""
+
+ // Historical fields must use their historical sequence/aggregation
options;
+ // latest options reference names that do not exist in the old schema.
+ [false, true].each { forceJni ->
+ sql """set force_jni_scanner=${forceJni}"""
+ "order_qt_sequence_latest_${forceJni}" """
+ SELECT id, current_sequence, payload FROM sequence_timeline
ORDER BY id
+ """
+ "order_qt_sequence_old_snapshot_${forceJni}" """
+ SELECT id, old_sequence, payload
+ FROM sequence_timeline FOR VERSION AS OF ${sequenceBase} ORDER
BY id
+ """
+ "order_qt_sequence_new_snapshot_${forceJni}" """
+ SELECT id, current_sequence, payload
+ FROM sequence_timeline FOR VERSION AS OF ${sequenceCurrent}
ORDER BY id
+ """
+ "order_qt_sequence_options_${forceJni}" """
+ SELECT id, old_sequence, payload
+ FROM
sequence_timeline@options('scan.snapshot-id'='${sequenceBase}')
+ ORDER BY id
+ """
+ "order_qt_sequence_tag_${forceJni}" """
+ SELECT id, old_sequence, payload
+ FROM sequence_timeline FOR VERSION AS OF 'sequence_base' ORDER
BY id
+ """
+ "order_qt_aggregation_latest_${forceJni}" """
+ SELECT id, current_total FROM aggregation_timeline ORDER BY id
+ """
+ "order_qt_aggregation_old_snapshot_${forceJni}" """
+ SELECT id, original_total
+ FROM aggregation_timeline FOR VERSION AS OF ${aggregationBase}
ORDER BY id
+ """
+ "order_qt_aggregation_intermediate_snapshot_${forceJni}" """
+ SELECT id, intermediate_total
+ FROM aggregation_timeline FOR VERSION AS OF
${aggregationIntermediate} ORDER BY id
+ """
+ "order_qt_aggregation_options_${forceJni}" """
+ SELECT id, original_total
+ FROM
aggregation_timeline@options('scan.snapshot-id'='${aggregationBase}')
+ ORDER BY id
+ """
+ "order_qt_aggregation_tag_${forceJni}" """
+ SELECT id, original_total
+ FROM aggregation_timeline@tag(aggregation_base) ORDER BY id
+ """
+ // Historical schema restoration must not revive a removed startup
mode.
+ "order_qt_removed_scan_mode_${forceJni}" """
+ SELECT id, value FROM scan_state_history
+ FOR VERSION AS OF ${scanStateSnapshot} ORDER BY id
+ """
+ }
+ } finally {
+ sql """set force_jni_scanner=false"""
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
