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 77626ad5f1d [fix](paimon) Bind REST schemas to stored table
generations (#69158)
77626ad5f1d is described below
commit 77626ad5f1d48d92a9a1be3ee1790310a0ec803f
Author: Gabriel <[email protected]>
AuthorDate: Fri Oct 16 15:42:26 2026 +0800
[fix](paimon) Bind REST schemas to stored table generations (#69158)
### What problem does this PR solve?
A normal SELECT through a Paimon REST catalog can fail with:
```
Paimon table generation changed after the statement was pinned; retry the
statement
```
For example, `test_paimon_rest_catalog_write` creates a table and
completes its INSERT statements, then fails while planning the first
SELECT. RESTCatalog reconstructs `TableSchema` from a REST `Schema`
using `TableSchema.create`. Its creation time, schema-format version and
highest allocated field ID need not match the physical schema. Comparing
these synthetic values to the stored schema falsely reports replacement.
Bind REST schemas to the physical schema at the same ID before caching.
To keep metadata and physical identity consistent during DROP/CREATE,
use the first REST response only to locate schema files, capture their
full identities, reload REST metadata without the Doris table cache, and
require the physical identities to remain unchanged. The comparison
includes stored options, branch/path and fallback shape. Then verify the
logical fields/keys/comment and retain the fresh REST options, FileIO
and catalog callbacks.
For example, replacing a primary-key table `(x, y)` with `bucket=2,
bucket-key=x` by one using `bucket-key=y` must never combine the
replacement schema identity with the original options: `WHERE x=1` could
otherwise prune buckets containing matching rows. A replacement before
the physical-read bracket uses the fresh response; a replacement during
it fails the load without caching it. Fallback leaves are bound and
checked independently.
Cache hits never rebind retained handles. Existing strict
schema/snapshot generation checks remain unchanged. Each REST table
cache miss now performs two REST table lookups and two physical schema
reads per leaf; cache hits add no I/O. Non-REST catalogs and non-file
tables retain their existing loading path.
### Validation
- Nine REST schema tests cover synthetic metadata, cache
enabled/disabled, null/same UUID, option/IO/environment preservation,
replacements before/during binding (including equal creation
timestamps), retry after failed loading, retained-handle rejection,
refresh, logical mismatch, changed location and fallback
replacement/shape changes.
- A real local Paimon write/read fixture reproduces missing rows with
mixed-generation bucket options and reads all 16 matching rows after the
fix. The original implementation failed the three initial race tests.
- `mvn -f fe/pom.xml -pl :fe-connector-paimon -am install
-Dmaven.build.cache.enabled=false` passed, including FE Checkstyle and
plugin zip assembly: 1,465 tests, zero failures/errors, one existing
live-connectivity test skipped. The Paimon module has 811 passing tests
and that one skip.
- The REST write regression also checks refresh/re-read consistency
after its existing golden assertion. Groovy syntax compilation and `git
diff --check` passed.
Tests use real local Paimon files and SDK readers/writers with a
recording REST catalog. A live REST/DLF service was not exercised
locally; the external regression remains for CI. The existing live test
skips when `PAIMON_WAREHOUSE` is not configured.
### Release note
Fix false table-generation-change errors and prevent mixed-generation
metadata when reading Paimon REST tables.
### Check List (For Author)
- [x] Unit Test
- [x] Regression test extension
- [x] Behavior changed: valid REST schemas are accepted; concurrent
replacement during binding is rejected
- [x] No documentation needed
---
.../connector/paimon/PaimonMetaCacheCatalog.java | 11 +-
.../doris/connector/paimon/PaimonSchemaPin.java | 64 ++++
.../connector/paimon/PaimonRestSchemaTest.java | 347 +++++++++++++++++++++
.../paimon/test_paimon_rest_catalog_write.groovy | 5 +
4 files changed, 425 insertions(+), 2 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalog.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalog.java
index b537b6f1e13..d82e67ec163 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalog.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonMetaCacheCatalog.java
@@ -45,6 +45,7 @@ import org.apache.paimon.utils.SegmentsCache;
import java.time.Duration;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
@@ -211,8 +212,14 @@ final class PaimonMetaCacheCatalog extends DelegateCatalog
{
try {
cached = tableCache.get(identifier, ignored -> {
try {
- Table loaded =
attachPerTableCaches(super.getTable(identifier));
- return new ExpiringValue<>(loaded,
nanoTime.getAsLong());
+ Table loaded = super.getTable(identifier);
+ if (loaded instanceof FileStoreTable
+ &&
PaimonCatalogOps.CatalogBackedPaimonCatalogOps.restCatalog(wrapped) != null) {
+ Map<String, String> schemas =
+
PaimonSchemaPin.captureRestSchemas((FileStoreTable) loaded);
+ loaded =
PaimonSchemaPin.bindRestSchema(super.getTable(identifier), schemas);
+ }
+ return new
ExpiringValue<>(attachPerTableCaches(loaded), nanoTime.getAsLong());
} catch (TableNotExistException e) {
throw new CatalogLoadException(e);
}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonSchemaPin.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonSchemaPin.java
index 7cca13244be..0b633f4c452 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonSchemaPin.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonSchemaPin.java
@@ -21,10 +21,12 @@ import
org.apache.doris.connector.spi.DorisConnectorException;
import com.google.common.hash.Hashing;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.DataTable;
import org.apache.paimon.table.DelegatedFileStoreTable;
import org.apache.paimon.table.FallbackReadFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.utils.SnapshotManager;
@@ -125,6 +127,68 @@ final class PaimonSchemaPin {
}
}
+ static Map<String, String> captureRestSchemas(FileStoreTable table) {
+ Map<String, String> schemas = new HashMap<>();
+ captureRestSchemas(table, "", schemas);
+ schemas.put("shape", shape(table));
+ return schemas;
+ }
+
+ private static void captureRestSchemas(FileStoreTable table, String path,
Map<String, String> schemas) {
+ if (table instanceof FallbackReadFileStoreTable) {
+ FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable)
table;
+ captureRestSchemas(pair.wrapped(), path, schemas);
+ captureRestSchemas(pair.other(), path + "fallback.", schemas);
+ } else if (table instanceof DelegatedFileStoreTable) {
+ captureRestSchemas(((DelegatedFileStoreTable) table).wrapped(),
path, schemas);
+ } else {
+ SchemaManager manager = table.schemaManager();
+ schemas.put(path, restSchemaIdentity(manager,
manager.schema(table.schema().id())));
+ }
+ }
+
+ static FileStoreTable bindRestSchema(Table refreshed, Map<String, String>
schemas) {
+ if (!(refreshed instanceof FileStoreTable) ||
!shape(refreshed).equals(schemas.get("shape"))) {
+ throw changed();
+ }
+ return bindRestSchema((FileStoreTable) refreshed, "", schemas);
+ }
+
+ private static FileStoreTable bindRestSchema(FileStoreTable table, String
path, Map<String, String> schemas) {
+ if (table instanceof FallbackReadFileStoreTable) {
+ FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable)
table;
+ return PaimonTableDecorators.rebuildFallbackPair(pair,
+ bindRestSchema(pair.wrapped(), path, schemas),
+ bindRestSchema(pair.other(), path + "fallback.", schemas));
+ }
+ if (table instanceof DelegatedFileStoreTable) {
+ return PaimonTableDecorators.replaceWrapped(table,
+ bindRestSchema(((DelegatedFileStoreTable)
table).wrapped(), path, schemas));
+ }
+ TableSchema retained = table.schema();
+ SchemaManager manager = table.schemaManager();
+ TableSchema persisted = manager.schema(retained.id());
+ // Bracket the fresh REST lookup with physical reads. The first REST
handle only locates files;
+ // reusing its options could combine an old bucket key with a
replacement's schema identity.
+ if (!restSchemaIdentity(manager, persisted).equals(schemas.get(path)))
{
+ throw changed();
+ }
+ // REST synthesizes timeMillis, version and highestFieldId, so compare
its logical schema only
+ // after confirming that the physical generation stayed unchanged
across the REST lookup.
+ if (!retained.copy(Collections.emptyMap()).toSchema()
+ .equals(persisted.copy(Collections.emptyMap()).toSchema())) {
+ throw new DorisConnectorException("Paimon REST schema does not
match stored schema "
+ + retained.id() + "; retry the statement");
+ }
+ // Preserve the freshly loaded REST FileIO, catalog callbacks and
options. Cache hits never rebind.
+ return table.copy(persisted.copy(retained.options()));
+ }
+
+ private static String restSchemaIdentity(SchemaManager manager,
TableSchema schema) {
+ // Include the branch/path and all persisted options, even if a
replacement reuses a timestamp.
+ return manager.toSchemaPath(schema.id()) + "\n" + schemaDigest(schema);
+ }
+
static void validateRetainedSchema(TableSchema retained, TableSchema
persisted) {
// Dynamic copies legitimately overlay options, but preserve the
schema's structural metadata
// and creation timestamp. Compare those against the same ID's
physical history, not latest.
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonRestSchemaTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonRestSchemaTest.java
new file mode 100644
index 00000000000..5ad0f5f52f0
--- /dev/null
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonRestSchemaTest.java
@@ -0,0 +1,347 @@
+// 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.cache.CatalogMetaCache;
+import org.apache.doris.connector.spi.DorisConnectorException;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+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.predicate.PredicateBuilder;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.rest.RESTCatalog;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+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.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.DataTypes;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+
+class PaimonRestSchemaTest {
+ private static final Identifier TABLE = Identifier.create("db", "events");
+
+ @TempDir
+ java.nio.file.Path directory;
+
+ @Test
+ void restSchemaCanBePinnedWithCacheEnabledOrDisabled() throws Exception {
+ for (boolean cacheEnabled : new boolean[] {true, false}) {
+ FileStoreTable physical = physicalTable("table-" + cacheEnabled,
"value", 1000L);
+ FileStoreTable restTable = restTable(physical, cacheEnabled ?
"rest-table-id" : null);
+ Assertions.assertNotEquals(physical.schema().timeMillis(),
restTable.schema().timeMillis());
+ Assertions.assertNotEquals(physical.schema().highestFieldId(),
restTable.schema().highestFieldId());
+ Assertions.assertNotEquals(physical.schema().version(),
restTable.schema().version());
+ AtomicReference<Table> response = new AtomicReference<>(restTable);
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged();
+ RESTCatalog rest = restCatalog(response, new
AtomicInteger())) {
+ Catalog cached = cached(rest, owner, cacheEnabled);
+ FileStoreTable loaded = (FileStoreTable)
cached.getTable(TABLE);
+ Map<String, String> pin = pin(loaded);
+ FileStoreTable scan =
PaimonScanParams.applyOptionsWithoutTimeTravel(loaded, pin);
+ PaimonSchemaPin.validateRetainedSchema(scan.schema(),
physical.schema());
+ Assertions.assertEquals("false",
scan.options().get("file-index.read.enabled"));
+ Assertions.assertSame(restTable.fileIO(), scan.fileIO());
+ Assertions.assertSame(restTable.catalogEnvironment(),
scan.catalogEnvironment());
+ }
+ }
+ }
+
+ @Test
+ void
cachedRestHandleRejectsSameShapeReplacementEvenWhenPinDescribesReplacement()
throws Exception {
+ FileStoreTable physical = physicalTable("replaced", "value", 1000L);
+ AtomicReference<Table> response = new
AtomicReference<>(restTable(physical));
+ AtomicInteger loads = new AtomicInteger();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged();
+ RESTCatalog rest = restCatalog(response, loads)) {
+ Catalog cached = cached(rest, owner, true);
+ FileStoreTable retained = (FileStoreTable) cached.getTable(TABLE);
+ physical.schemaManager().deleteSchema(0L);
+ FileStoreTable replacement = physicalTable("replaced", "value",
2000L);
+ response.set(restTable(replacement));
+ Assertions.assertSame(retained, cached.getTable(TABLE));
+ Assertions.assertEquals(2, loads.get());
+ Map<String, String> pin = pin(retained);
+ Assertions.assertThrows(DorisConnectorException.class,
+ () ->
PaimonScanParams.applyOptionsWithoutTimeTravel(retained, pin));
+ cached.invalidateTable(TABLE);
+ FileStoreTable fresh = (FileStoreTable) cached.getTable(TABLE);
+ Assertions.assertDoesNotThrow(() ->
PaimonScanParams.applyOptionsWithoutTimeTravel(fresh, pin));
+ Assertions.assertEquals(4, loads.get());
+ }
+ }
+
+ @Test
+ void restAndPhysicalFieldsMustAgreeBeforeBinding() throws Exception {
+ FileStoreTable physical = physicalTable("mismatch", "old_value",
1000L);
+ FileStoreTable staleRest = restTable(physical);
+ physical.schemaManager().deleteSchema(0L);
+ physicalTable("mismatch", "new_value", 2000L);
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged();
+ RESTCatalog rest = restCatalog(new
AtomicReference<>(staleRest), new AtomicInteger())) {
+ Catalog cached = cached(rest, owner, true);
+ Assertions.assertThrows(DorisConnectorException.class, () ->
cached.getTable(TABLE));
+ }
+ }
+
+ @Test
+ void fallbackBranchesKeepTheirOwnPhysicalSchemas() throws Exception {
+ FileStoreTable main = physicalTable("main", "value", 1000L);
+ FileStoreTable fallback = physicalTable("fallback", "value", 2000L);
+ FallbackReadFileStoreTable pair = new
FallbackReadFileStoreTable(restTable(main), restTable(fallback), true);
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged();
+ RESTCatalog rest = restCatalog(new AtomicReference<>(pair),
new AtomicInteger())) {
+ FallbackReadFileStoreTable loaded = (FallbackReadFileStoreTable)
cached(rest, owner, true).getTable(TABLE);
+ PaimonSchemaPin.validateRetainedSchema(loaded.wrapped().schema(),
main.schema());
+ PaimonSchemaPin.validateRetainedSchema(loaded.other().schema(),
fallback.schema());
+ Assertions.assertSame(pair.wrapped().catalogEnvironment(),
loaded.wrapped().catalogEnvironment());
+ Assertions.assertSame(pair.other().catalogEnvironment(),
loaded.other().catalogEnvironment());
+ Assertions.assertTrue(PaimonReaderOptions.isWrappedFirst(loaded));
+ }
+ }
+
+ @Test
+ void replacementBeforeFirstSchemaReadUsesFreshRestOptions() throws
Exception {
+ FileStoreTable original = bucketTable("before-read");
+ FileStoreTable replacement = replacement(original, 2000L);
+ FileStoreTable oldRest = restTable(original);
+ FileStoreTable newRest = restTable(replacement);
+ AtomicInteger loads = new AtomicInteger();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged();
+ RESTCatalog rest = restCatalog(() -> {
+ if (loads.get() == 1) {
+ replaceSchema(replacement);
+ writeBucketRows(replacement);
+ return oldRest;
+ }
+ return newRest;
+ }, loads)) {
+ FileStoreTable loaded = (FileStoreTable) cached(rest, owner,
true).getTable(TABLE);
+ FileStoreTable mixed =
oldRest.copy(replacement.schema().copy(oldRest.options()));
+ Assertions.assertTrue(matchingRows(mixed).size() < 16,
+ "the fixture must expose bucket pruning with
mixed-generation options");
+ Assertions.assertEquals(16, matchingRows(loaded).size(),
+ "WHERE x = 1 must retain rows from every replacement
bucket keyed by y");
+ Assertions.assertEquals("y", loaded.options().get("bucket-key"),
+ "a replacement schema must never be combined with the
original bucket key");
+ PaimonSchemaPin.validateRetainedSchema(loaded.schema(),
replacement.schema());
+ }
+ }
+
+ @Test
+ void replacementDuringRestReloadRejectsTheLoadWithoutCachingIt() throws
Exception {
+ for (long replacementTime : new long[] {1000L, 2000L}) {
+ FileStoreTable original = bucketTable("during-reload-" +
replacementTime);
+ FileStoreTable replacement = replacement(original,
replacementTime);
+ FileStoreTable oldRest = restTable(original);
+ FileStoreTable newRest = restTable(replacement);
+ AtomicInteger loads = new AtomicInteger();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged();
+ RESTCatalog rest = restCatalog(() -> {
+ if (loads.get() == 2) {
+ replaceSchema(replacement);
+ }
+ return loads.get() <= 2 ? oldRest : newRest;
+ }, loads)) {
+ Catalog cached = cached(rest, owner, true);
+ Assertions.assertThrows(DorisConnectorException.class, () ->
cached.getTable(TABLE));
+ FileStoreTable retried = (FileStoreTable)
cached.getTable(TABLE);
+ Assertions.assertEquals("y",
retried.options().get("bucket-key"));
+ PaimonSchemaPin.validateRetainedSchema(retried.schema(),
replacement.schema());
+ }
+ }
+ }
+
+ @Test
+ void fallbackReplacementDuringRestReloadIsAlsoRejected() throws Exception {
+ FileStoreTable main = bucketTable("stable-main");
+ FileStoreTable fallback = bucketTable("changing-fallback");
+ FileStoreTable replacement = replacement(fallback, 2000L);
+ FallbackReadFileStoreTable pair = new
FallbackReadFileStoreTable(restTable(main), restTable(fallback), true);
+ AtomicInteger loads = new AtomicInteger();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged();
+ RESTCatalog rest = restCatalog(() -> {
+ if (loads.get() == 2) {
+ replaceSchema(replacement);
+ }
+ return pair;
+ }, loads)) {
+ Catalog cached = cached(rest, owner, true);
+ Assertions.assertThrows(DorisConnectorException.class, () ->
cached.getTable(TABLE));
+ }
+ }
+
+ @Test
+ void fallbackShapeChangeDuringReloadIsRejected() throws Exception {
+ FileStoreTable main = restTable(bucketTable("shape-main"));
+ FileStoreTable fallback = restTable(bucketTable("shape-fallback"));
+ AtomicInteger loads = new AtomicInteger();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged();
+ RESTCatalog rest = restCatalog(() -> loads.get() == 1
+ ? new FallbackReadFileStoreTable(main, fallback, true)
: main, loads)) {
+ Assertions.assertThrows(DorisConnectorException.class, () ->
cached(rest, owner, true).getTable(TABLE));
+ }
+ }
+
+ @Test
+ void changedLocationWithIdenticalSchemaDuringReloadIsRejected() throws
Exception {
+ FileStoreTable original = restTable(bucketTable("original-location"));
+ FileStoreTable relocated = restTable(bucketTable("new-location"));
+ AtomicInteger loads = new AtomicInteger();
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged();
+ RESTCatalog rest = restCatalog(() -> loads.get() == 1 ?
original : relocated, loads)) {
+ Assertions.assertThrows(DorisConnectorException.class, () ->
cached(rest, owner, true).getTable(TABLE));
+ }
+ }
+
+ private static void writeBucketRows(FileStoreTable table) {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite(); BatchTableCommit
commit = builder.newCommit()) {
+ for (int y = 0; y < 16; y++) {
+ write.write(GenericRow.of(1, y));
+ }
+ commit.commit(write.prepareCommit());
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static List<Integer> matchingRows(FileStoreTable table) throws
Exception {
+ ReadBuilder read = table.newReadBuilder().withFilter(new
PredicateBuilder(table.rowType()).equal(0, 1));
+ List<Integer> values = new ArrayList<>();
+ for (Split split : read.newScan().plan().splits()) {
+ try (RecordReader<InternalRow> reader =
read.newRead().createReader(split)) {
+ reader.forEachRemaining(row -> values.add(row.getInt(1)));
+ }
+ }
+ return values;
+ }
+
+ private FileStoreTable bucketTable(String name) throws Exception {
+ LocalFileIO io = LocalFileIO.create();
+ org.apache.paimon.fs.Path location = new
org.apache.paimon.fs.Path(directory.resolve(name).toUri());
+ Schema schema = Schema.newBuilder().column("x",
DataTypes.INT().notNull())
+ .column("y", DataTypes.INT().notNull()).primaryKey("x", "y")
+ .option("bucket", "2").option("bucket-key", "x").build();
+ TableSchema fields = TableSchema.create(0L, schema);
+ TableSchema persisted = new TableSchema(fields.version(), 0L,
fields.fields(), fields.highestFieldId(),
+ fields.partitionKeys(), fields.primaryKeys(),
fields.options(), fields.comment(), 1000L);
+ Assertions.assertTrue(new SchemaManager(io,
location).commit(persisted));
+ return FileStoreTableFactory.create(io, location, persisted);
+ }
+
+ private static FileStoreTable replacement(FileStoreTable original, long
time) {
+ TableSchema schema = original.schema();
+ Map<String, String> options = new HashMap<>(schema.options());
+ options.put("bucket-key", "y");
+ TableSchema replaced = new TableSchema(schema.version(), schema.id(),
schema.fields(), schema.highestFieldId(),
+ schema.partitionKeys(), schema.primaryKeys(), options,
schema.comment(), time);
+ return FileStoreTableFactory.create(original.fileIO(),
original.location(), replaced);
+ }
+
+ private static void replaceSchema(FileStoreTable replacement) {
+ try {
+
replacement.fileIO().overwriteFileUtf8(replacement.schemaManager().toSchemaPath(replacement.schema().id()),
+ replacement.schema().toString());
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private FileStoreTable physicalTable(String name, String valueColumn, long
creationTime) throws Exception {
+ LocalFileIO io = LocalFileIO.create();
+ org.apache.paimon.fs.Path location = new
org.apache.paimon.fs.Path(directory.resolve(name).toUri());
+ Schema schema = Schema.newBuilder().column("id", DataTypes.INT())
+ .column(valueColumn, DataTypes.STRING()).option("bucket",
"-1").build();
+ TableSchema fields = TableSchema.create(0L, schema);
+ // Older schema formats and removed field IDs are also absent from
REST's Schema payload.
+ TableSchema persisted = new TableSchema(TableSchema.PAIMON_07_VERSION,
0L, fields.fields(), 10,
+ fields.partitionKeys(), fields.primaryKeys(),
fields.options(), fields.comment(), creationTime);
+ Assertions.assertTrue(new SchemaManager(io,
location).commit(persisted));
+ return FileStoreTableFactory.create(io, location, persisted);
+ }
+
+ private static FileStoreTable restTable(FileStoreTable physical) {
+ return restTable(physical, "rest-table-id");
+ }
+
+ private static FileStoreTable restTable(FileStoreTable physical, String
uuid) {
+ TableSchema reconstructed = TableSchema.create(physical.schema().id(),
physical.schema().toSchema());
+ Map<String, String> options = new HashMap<>(reconstructed.options());
+ options.put("file-index.read.enabled", "false");
+ return FileStoreTableFactory.create(physical.fileIO(),
physical.location(), reconstructed.copy(options),
+ new CatalogEnvironment(TABLE, uuid, null, null, null, null,
false, true));
+ }
+
+ private static Map<String, String> pin(FileStoreTable table) {
+ Map<String, String> options = PaimonScanParams.withBoundSchema(
+ PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(),
-1L), table.schema().id());
+ options.putAll(PaimonSchemaPin.capture(table, table.schema().id(),
-1L));
+ return options;
+ }
+
+ private static Catalog cached(RESTCatalog rest, CatalogMetaCache owner,
boolean enabled) {
+ return PaimonMetaCacheCatalog.tryToCreate(rest, owner, 100, 100, new
Options(),
+ enabled, true, catalog -> catalog);
+ }
+
+ private static RESTCatalog restCatalog(AtomicReference<Table> response,
AtomicInteger loads) {
+ return restCatalog(response::get, loads);
+ }
+
+ private static RESTCatalog restCatalog(Supplier<Table> response,
AtomicInteger loads) {
+ Options options = new Options();
+ options.set("uri", "http://localhost:1";);
+ options.set("prefix", "test-prefix");
+ options.set("token.provider", "bear");
+ options.set("token", "test-token");
+ return new RESTCatalog(CatalogContext.create(options), false) {
+ @Override
+ public Table getTable(Identifier identifier) {
+ loads.incrementAndGet();
+ return response.get();
+ }
+ };
+ }
+}
diff --git
a/regression-test/suites/external_table_p2/paimon/test_paimon_rest_catalog_write.groovy
b/regression-test/suites/external_table_p2/paimon/test_paimon_rest_catalog_write.groovy
index 40f04e5b48a..f316089b0cc 100644
---
a/regression-test/suites/external_table_p2/paimon/test_paimon_rest_catalog_write.groovy
+++
b/regression-test/suites/external_table_p2/paimon/test_paimon_rest_catalog_write.groovy
@@ -73,6 +73,11 @@ suite("test_paimon_rest_catalog_write", "p2,external") {
ORDER BY id
"""
+ // REST reloads reconstruct schema metadata; an unchanged table must keep
the same readable rows.
+ def appendRows = sql """SELECT id, name, score, region FROM
`${appendTable}` ORDER BY id"""
+ sql """REFRESH TABLE `${appendTable}`"""
+ assertEquals(appendRows, sql("""SELECT id, name, score, region FROM
`${appendTable}` ORDER BY id"""))
+
sql """
CREATE TABLE `${primaryKeyTable}` (
user_id INT NOT NULL,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
