diff --git a/paimon-api/src/main/java/org/apache/paimon/table/TableDescriptor.java b/paimon-api/src/main/java/org/apache/paimon/table/TableDescriptor.java new file mode 100644 index 000000000000..20e490ec5123 --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/table/TableDescriptor.java @@ -0,0 +1,158 @@ +/* + * 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.paimon.table; + +import org.apache.paimon.annotation.Experimental; +import org.apache.paimon.schema.TableSchema; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty; + +import javax.annotation.Nullable; + +/** + * A portable, versioned description of a Paimon table, sufficient for a non-Java runtime (e.g. a + * paimon-rust / paimon-cpp reader embedded in a query engine) to rebuild a table for scan/read + * without a catalog lookup. + * + *

It carries only what a reader needs: the table {@link #path}, the full {@link TableSchema} + * (identical to an on-disk {@code schema/schema-N} file), and a few optional hints. It deliberately + * excludes JVM-only state (e.g. {@code FileIO}, {@code CatalogEnvironment}) so the JSON wire form + * is language-neutral. Serialize/deserialize it through {@code JsonSerdeUtil} (which registers the + * custom {@code TableSchema} serde) — not a bare Jackson {@code ObjectMapper}. + * + *

The descriptor's own {@link #version} is distinct from the nested {@link TableSchema}'s + * schema-format version. + * + *

This is a passive data holder: it performs no validation on construction or parse. Validation + * lives at the producer ({@code TableDescriptorSerializer}, which rejects unsupported table shapes) + * and at the reader (which fails fast on an unknown version). + */ +@JsonIgnoreProperties(ignoreUnknown = true) +@JsonInclude(JsonInclude.Include.NON_NULL) +@Experimental +public class TableDescriptor { + + /** + * Current descriptor contract version. A reader that does not recognize the version must fail + * fast rather than guess. + */ + public static final int CURRENT_VERSION = 1; + + private static final String FIELD_VERSION = "version"; + private static final String FIELD_PATH = "path"; + private static final String FIELD_TABLE_SCHEMA = "tableSchema"; + private static final String FIELD_DATABASE = "database"; + private static final String FIELD_NAME = "name"; + private static final String FIELD_BRANCH = "branch"; + private static final String FIELD_SNAPSHOT_ID = "snapshotId"; + + @JsonProperty(FIELD_VERSION) + private final int version; + + @JsonProperty(FIELD_PATH) + private final String path; + + @JsonProperty(FIELD_TABLE_SCHEMA) + private final TableSchema tableSchema; + + @JsonProperty(FIELD_DATABASE) + @Nullable + private final String database; + + @JsonProperty(FIELD_NAME) + @Nullable + private final String name; + + /** The table branch. Omitted (null) for the main branch. */ + @JsonProperty(FIELD_BRANCH) + @Nullable + private final String branch; + + /** + * The snapshot resolved by the planner when time travel was requested ({@code scan.snapshot-id} + * / {@code scan.tag-name} / {@code scan.timestamp-millis} / {@code scan.version} / ...). + * Omitted (null) for an ordinary scan, in which case a reader that plans from this descriptor + * reads the latest snapshot. Carrying the resolved snapshot id (rather than the raw selectors) + * keeps the contract engine- and version-neutral: the reader just pins this snapshot. + */ + @JsonProperty(FIELD_SNAPSHOT_ID) + @Nullable + private final Long snapshotId; + + @JsonCreator + public TableDescriptor( + @JsonProperty(FIELD_VERSION) int version, + @JsonProperty(FIELD_PATH) String path, + @JsonProperty(FIELD_TABLE_SCHEMA) TableSchema tableSchema, + @JsonProperty(FIELD_DATABASE) @Nullable String database, + @JsonProperty(FIELD_NAME) @Nullable String name, + @JsonProperty(FIELD_BRANCH) @Nullable String branch, + @JsonProperty(FIELD_SNAPSHOT_ID) @Nullable Long snapshotId) { + this.version = version; + this.path = path; + this.tableSchema = tableSchema; + this.database = database; + this.name = name; + this.branch = branch; + this.snapshotId = snapshotId; + } + + @JsonGetter(FIELD_VERSION) + public int version() { + return version; + } + + @JsonGetter(FIELD_PATH) + public String path() { + return path; + } + + @JsonGetter(FIELD_TABLE_SCHEMA) + public TableSchema tableSchema() { + return tableSchema; + } + + @JsonGetter(FIELD_DATABASE) + @Nullable + public String database() { + return database; + } + + @JsonGetter(FIELD_NAME) + @Nullable + public String name() { + return name; + } + + @JsonGetter(FIELD_BRANCH) + @Nullable + public String branch() { + return branch; + } + + @JsonGetter(FIELD_SNAPSHOT_ID) + @Nullable + public Long snapshotId() { + return snapshotId; + } +} diff --git a/paimon-api/src/test/java/org/apache/paimon/table/TableDescriptorTest.java b/paimon-api/src/test/java/org/apache/paimon/table/TableDescriptorTest.java new file mode 100644 index 000000000000..59dff437e885 --- /dev/null +++ b/paimon-api/src/test/java/org/apache/paimon/table/TableDescriptorTest.java @@ -0,0 +1,150 @@ +/* + * 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.paimon.table; + +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.IntType; +import org.apache.paimon.utils.JsonSerdeUtil; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link TableDescriptor} JSON serialization. */ +public class TableDescriptorTest { + + private static TableSchema tableSchema() { + List fields = + Arrays.asList( + new DataField(0, "f0", new IntType()), + new DataField(1, "f1", new IntType())); + Schema schema = + new Schema( + fields, + Collections.singletonList("f0"), + Arrays.asList("f0", "f1"), + Collections.singletonMap("bucket", "1"), + "comment"); + return TableSchema.create(5, schema); + } + + @Test + public void testJsonRoundTrip() { + TableDescriptor descriptor = + new TableDescriptor( + TableDescriptor.CURRENT_VERSION, + "file:///tmp/paimon/mydb.db/t", + tableSchema(), + "mydb", + "t", + null, + null); + + String json = JsonSerdeUtil.toJson(descriptor); + TableDescriptor parsed = JsonSerdeUtil.fromJson(json, TableDescriptor.class); + + assertThat(parsed.version()).isEqualTo(1); + assertThat(parsed.path()).isEqualTo("file:///tmp/paimon/mydb.db/t"); + assertThat(parsed.database()).isEqualTo("mydb"); + assertThat(parsed.name()).isEqualTo("t"); + assertThat(parsed.branch()).isNull(); + assertThat(parsed.tableSchema().id()).isEqualTo(5); + assertThat(parsed.tableSchema().primaryKeys()).containsExactly("f0", "f1"); + assertThat(parsed.tableSchema().partitionKeys()).containsExactly("f0"); + assertThat(parsed.tableSchema().options()).containsEntry("bucket", "1"); + } + + @Test + public void testBranchRoundTrip() { + TableDescriptor descriptor = + new TableDescriptor( + TableDescriptor.CURRENT_VERSION, + "file:///tmp/paimon/mydb.db/t", + tableSchema(), + "mydb", + "t", + "b1", + null); + + String json = JsonSerdeUtil.toJson(descriptor); + assertThat(json).contains("\"branch\""); + TableDescriptor parsed = JsonSerdeUtil.fromJson(json, TableDescriptor.class); + assertThat(parsed.branch()).isEqualTo("b1"); + } + + @Test + public void testSnapshotIdRoundTrip() { + TableDescriptor descriptor = + new TableDescriptor( + TableDescriptor.CURRENT_VERSION, + "file:///tmp/paimon/mydb.db/t", + tableSchema(), + "mydb", + "t", + null, + 7L); + + String json = JsonSerdeUtil.toJson(descriptor); + assertThat(json).contains("\"snapshotId\""); + TableDescriptor parsed = JsonSerdeUtil.fromJson(json, TableDescriptor.class); + assertThat(parsed.snapshotId()).isEqualTo(7L); + } + + @Test + public void testNestedTableSchemaUsesPaimonSerde() { + // The nested tableSchema must round-trip through Paimon's custom serde + // (SchemaSerializer), preserving id/highestFieldId losslessly. + TableDescriptor descriptor = + new TableDescriptor( + TableDescriptor.CURRENT_VERSION, + "file:///tmp/t", + tableSchema(), + null, + null, + null, + null); + + String json = JsonSerdeUtil.toJson(descriptor); + assertThat(json).contains("\"tableSchema\"").contains("\"highestFieldId\""); + + TableDescriptor parsed = JsonSerdeUtil.fromJson(json, TableDescriptor.class); + assertThat(parsed.tableSchema().highestFieldId()) + .isEqualTo(descriptor.tableSchema().highestFieldId()); + assertThat(parsed.database()).isNull(); + } + + @Test + public void testUnknownFieldsIgnored() { + // Forward compatibility: unknown top-level fields must not fail parsing. + String json = + "{\"version\":1,\"path\":\"file:///tmp/t\",\"futureField\":\"x\"," + + "\"tableSchema\":" + + JsonSerdeUtil.toJson(tableSchema()) + + "}"; + TableDescriptor parsed = JsonSerdeUtil.fromJson(json, TableDescriptor.class); + assertThat(parsed.version()).isEqualTo(1); + assertThat(parsed.path()).isEqualTo("file:///tmp/t"); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/table/TableDescriptorSerializer.java b/paimon-core/src/main/java/org/apache/paimon/table/TableDescriptorSerializer.java new file mode 100644 index 000000000000..ab3575d99d5c --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/TableDescriptorSerializer.java @@ -0,0 +1,118 @@ +/* + * 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.paimon.table; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.Snapshot; +import org.apache.paimon.annotation.Experimental; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.source.snapshot.TimeTravelUtil; +import org.apache.paimon.utils.JsonSerdeUtil; + +import javax.annotation.Nullable; + +import java.util.Optional; + +/** + * Builds a portable {@link TableDescriptor} from a resolved {@link FileStoreTable}, and serializes + * it to the cross-language JSON wire form. + * + *

A query engine that has already resolved a table for planning can ship this descriptor to a + * non-Java reader (paimon-rust / paimon-cpp) so the reader rebuilds the table without re-resolving + * it through a catalog. Only the reader-relevant metadata is captured — the table {@link + * org.apache.paimon.table.DataTable#location() location} and its persisted schema; JVM-only + * state, dynamic session options (including credentials), and the injected path option are + * intentionally dropped. The resolved branch and, when time travel was requested, the resolved + * snapshot id are carried explicitly. + * + *

Snapshot selection has two modes: if a reader consumes a co-shipped planner-provided split, + * that split pins the exact files/snapshot; if a reader instead plans from this descriptor, it must + * pin {@link TableDescriptor#snapshotId()} when present to reproduce the resolved snapshot. + * + *

v1 supports plain file-store tables, including a specific (non-main) branch. Composite / + * delegating tables (fallback-branch, chain, privileged) are rejected: they delegate {@code + * location()}/{@code schema()} to their primary and would be serialized as a plain table, silently + * losing their merge / fallback semantics. + */ +@Experimental +public final class TableDescriptorSerializer { + + private TableDescriptorSerializer() {} + + /** + * Build a descriptor without database/table names (identifier is synthesized by the reader). + */ + public static TableDescriptor from(FileStoreTable table) { + return from(table, null, null); + } + + /** + * Build a descriptor, optionally carrying the database and table names for better reader-side + * identifiers and diagnostics. + */ + public static TableDescriptor from( + FileStoreTable table, @Nullable String database, @Nullable String name) { + if (table instanceof DelegatedFileStoreTable) { + throw new UnsupportedOperationException( + "TableDescriptor v1 only supports plain file-store tables, got: " + + table.getClass().getName()); + } + if ((database == null) != (name == null)) { + throw new IllegalArgumentException( + "TableDescriptor must set both database and name, or neither"); + } + // Branch is derived from the resolved table options (the persisted schema does not record + // it). Omit "main" to keep the descriptor slim (absent == main). + String branch = CoreOptions.branch(table.schema().options()); + String descriptorBranch = Identifier.DEFAULT_MAIN_BRANCH.equals(branch) ? null : branch; + // If the planner requested time travel, ship the RESOLVED snapshot id (not the raw + // scan.* selectors): the persisted schema strips those dynamic options, and carrying the + // resolved result keeps the contract engine-/version-neutral (the reader just pins this + // snapshot). Ordinary scans resolve to empty -> null (reader reads latest / the split pins + // files). + Optional travelSnapshot = TimeTravelUtil.tryTravelToSnapshot(table); + Long snapshotId = travelSnapshot.map(Snapshot::id).orElse(null); + // Serialize the PERSISTED schema (schema-N), not table.schema(): the latter has dynamic + // options merged in by AbstractFileStoreTable.copyInternal, which can include credentials + // (e.g. fs.s3a.secret.key) and the injected path option. Storage options travel separately + // (the reader's FFI takes them out of band). schemaManager() is branch-scoped, so for a + // branch table this reads the branch's own persisted schema. + TableSchema schema = table.schemaManager().schema(table.schema().id()); + return new TableDescriptor( + TableDescriptor.CURRENT_VERSION, + table.location().toString(), + schema, + database, + name, + descriptorBranch, + snapshotId); + } + + /** Serialize a descriptor to its JSON wire form via Paimon's {@link JsonSerdeUtil}. */ + public static String toJson(TableDescriptor descriptor) { + return JsonSerdeUtil.toJson(descriptor); + } + + /** Convenience: {@link #from(FileStoreTable, String, String)} then {@link #toJson}. */ + public static String serialize( + FileStoreTable table, @Nullable String database, @Nullable String name) { + return toJson(from(table, database, name)); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/TableDescriptorGoldenTest.java b/paimon-core/src/test/java/org/apache/paimon/table/TableDescriptorGoldenTest.java new file mode 100644 index 000000000000..ac227dc2eed0 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/TableDescriptorGoldenTest.java @@ -0,0 +1,102 @@ +/* + * 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.paimon.table; + +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.IntType; +import org.apache.paimon.types.VarCharType; +import org.apache.paimon.utils.JsonSerdeUtil; + +import org.junit.jupiter.api.Test; + +import java.io.InputStream; +import java.util.Arrays; +import java.util.Collections; +import java.util.Scanner; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Pins the canonical cross-language {@link TableDescriptor} JSON. The checked-in golden {@code + * table/table-descriptor-v1.json} is the exact wire form the paimon-rust reader is tested against, + * so a drift on either side is caught here and in the Rust golden test. + */ +public class TableDescriptorGoldenTest { + + private static final String GOLDEN = "/table/table-descriptor-v1.json"; + + /** A fully deterministic descriptor (fixed schema id / highestFieldId / timeMillis). */ + static TableDescriptor canonicalDescriptor() { + TableSchema tableSchema = + new TableSchema( + 3, + 1L, + Arrays.asList( + new DataField(0, "id", new IntType(false)), + new DataField(1, "name", new VarCharType(true, 100)), + new DataField(2, "tags", new ArrayType(true, new IntType(true)))), + 2, + Collections.emptyList(), + Collections.singletonList("id"), + Collections.singletonMap("bucket", "1"), + "canonical", + 1700000000000L); + return new TableDescriptor( + TableDescriptor.CURRENT_VERSION, + "file:///tmp/warehouse/mydb.db/t", + tableSchema, + "mydb", + "t", + null, + null); + } + + private static String readGolden() { + try (InputStream in = TableDescriptorGoldenTest.class.getResourceAsStream(GOLDEN)) { + assertThat(in).as("golden resource %s must exist", GOLDEN).isNotNull(); + try (Scanner scanner = new Scanner(in, "UTF-8").useDelimiter("\\A")) { + return scanner.hasNext() ? scanner.next() : ""; + } + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + @Test + public void producesCanonicalGolden() { + String json = TableDescriptorSerializer.toJson(canonicalDescriptor()); + assertThat(json.trim()).isEqualTo(readGolden().trim()); + } + + @Test + public void parsesCanonicalGolden() { + TableDescriptor descriptor = JsonSerdeUtil.fromJson(readGolden(), TableDescriptor.class); + assertThat(descriptor.version()).isEqualTo(1); + assertThat(descriptor.path()).isEqualTo("file:///tmp/warehouse/mydb.db/t"); + assertThat(descriptor.database()).isEqualTo("mydb"); + assertThat(descriptor.name()).isEqualTo("t"); + assertThat(descriptor.tableSchema().id()).isEqualTo(1L); + assertThat(descriptor.tableSchema().highestFieldId()).isEqualTo(2); + assertThat(descriptor.tableSchema().primaryKeys()).containsExactly("id"); + assertThat(descriptor.tableSchema().fields()).hasSize(3); + assertThat(descriptor.tableSchema().fields().get(2).name()).isEqualTo("tags"); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/TableDescriptorSerializerTest.java b/paimon-core/src/test/java/org/apache/paimon/table/TableDescriptorSerializerTest.java new file mode 100644 index 000000000000..809a0639edd6 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/TableDescriptorSerializerTest.java @@ -0,0 +1,212 @@ +/* + * 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.paimon.table; + +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.IntType; +import org.apache.paimon.utils.JsonSerdeUtil; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.util.Arrays; +import java.util.Collections; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; + +/** Tests for {@link TableDescriptorSerializer}. */ +public class TableDescriptorSerializerTest { + + @TempDir java.nio.file.Path tempDir; + + private FileStoreTable createTable() { + return createTable(Collections.singletonMap("bucket", "2")); + } + + /** + * Create a real table with schema-0 persisted on disk, so {@code schemaManager().schema(id)} + * resolves (the serializer reads the persisted schema). + */ + private FileStoreTable createTable(Map options) { + Path tablePath = new Path(tempDir.toString()); + LocalFileIO fileIO = LocalFileIO.create(); + Schema schema = + new Schema( + Arrays.asList( + new DataField(0, "f0", new IntType()), + new DataField(1, "f1", new IntType())), + Collections.emptyList(), + Arrays.asList("f0", "f1"), + options, + "comment"); + try { + new SchemaManager(fileIO, tablePath).createTable(schema); + } catch (Exception e) { + throw new RuntimeException(e); + } + return FileStoreTableFactory.create(fileIO, tablePath); + } + + @Test + public void testFromFileStoreTable() { + FileStoreTable table = createTable(); + TableDescriptor descriptor = TableDescriptorSerializer.from(table, "db", "t"); + + assertThat(descriptor.version()).isEqualTo(TableDescriptor.CURRENT_VERSION); + assertThat(descriptor.path()).isEqualTo(table.location().toString()); + assertThat(descriptor.tableSchema().id()).isEqualTo(table.schema().id()); + assertThat(descriptor.database()).isEqualTo("db"); + assertThat(descriptor.name()).isEqualTo("t"); + } + + @Test + public void testFromWithoutNames() { + FileStoreTable table = createTable(); + TableDescriptor descriptor = TableDescriptorSerializer.from(table); + assertThat(descriptor.database()).isNull(); + assertThat(descriptor.name()).isNull(); + assertThat(descriptor.path()).isEqualTo(table.location().toString()); + } + + @Test + public void testSerializeRoundTripsThroughRustCompatibleJson() { + FileStoreTable table = createTable(); + String json = TableDescriptorSerializer.serialize(table, "db", "t"); + + TableDescriptor parsed = JsonSerdeUtil.fromJson(json, TableDescriptor.class); + assertThat(parsed.version()).isEqualTo(TableDescriptor.CURRENT_VERSION); + assertThat(parsed.path()).isEqualTo(table.location().toString()); + assertThat(parsed.tableSchema().id()).isEqualTo(table.schema().id()); + assertThat(parsed.tableSchema().primaryKeys()).containsExactly("f0", "f1"); + assertThat(parsed.tableSchema().options()).containsEntry("bucket", "2"); + } + + @Test + public void testRejectsPartialIdentifier() { + FileStoreTable table = createTable(); + assertThatThrownBy(() -> TableDescriptorSerializer.from(table, "db", null)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> TableDescriptorSerializer.from(table, null, "t")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testRejectsCompositeTable() { + // Delegating/composite tables (fallback / chain / privileged) delegate location()/schema() + // to their primary, so they would serialize as a plain table and silently drop their + // merge/fallback semantics. v1 must reject them. + assertThatThrownBy( + () -> + TableDescriptorSerializer.from( + mock(FallbackReadFileStoreTable.class))) + .isInstanceOf(UnsupportedOperationException.class); + } + + @Test + public void testDoesNotLeakDynamicCredentials() { + // table.schema() would carry dynamic session options (incl. credentials); the serializer + // must ship the persisted schema instead. + FileStoreTable table = + createTable().copy(Collections.singletonMap("fs.s3a.secret.key", "review-secret")); + + String json = TableDescriptorSerializer.serialize(table, "db", "t"); + assertThat(json).doesNotContain("review-secret"); + + TableDescriptor descriptor = TableDescriptorSerializer.from(table, "db", "t"); + assertThat(descriptor.tableSchema().options()).doesNotContainKey("fs.s3a.secret.key"); + // The persisted schema also drops the runtime-injected path option. + assertThat(descriptor.tableSchema().options()).doesNotContainKey("path"); + } + + @Test + public void testCarriesNonMainBranch() { + // A non-main branch (of a plain table) is carried so the reader reads the right branch. + // branch lives only in the resolved table's options, never in the persisted schema, so it + // is derived from the table; the persisted schema is read (branch-scoped) from the branch. + Path tablePath = new Path(tempDir.toString()); + LocalFileIO fileIO = LocalFileIO.create(); + Schema schema = + new Schema( + Arrays.asList( + new DataField(0, "f0", new IntType()), + new DataField(1, "f1", new IntType())), + Collections.emptyList(), + Arrays.asList("f0", "f1"), + Collections.singletonMap("bucket", "2"), + "comment"); + try { + new SchemaManager(fileIO, tablePath).createTable(schema); + new SchemaManager(fileIO, tablePath, "b1").createTable(schema); + } catch (Exception e) { + throw new RuntimeException(e); + } + FileStoreTable table = + FileStoreTableFactory.create(fileIO, tablePath) + .copy(Collections.singletonMap("branch", "b1")); + + TableDescriptor descriptor = TableDescriptorSerializer.from(table, "db", "t"); + assertThat(descriptor.branch()).isEqualTo("b1"); + // path stays the table root (branch-independent); branch travels in its own field. + assertThat(descriptor.path()).isEqualTo(table.location().toString()); + assertThat(descriptor.tableSchema().options()).doesNotContainKey("branch"); + } + + @Test + public void testCarriesResolvedSnapshotIdForTimeTravel() { + // A time-travel request (scan.snapshot-id/tag/timestamp/...) is resolved to a concrete + // snapshot at serialize time and shipped as snapshotId, so a reader planning from the + // descriptor reproduces exactly that snapshot instead of reading latest. + FileStoreTable table = createTable(); + writeOneRow(table); // creates snapshot 1 + + FileStoreTable travel = table.copy(Collections.singletonMap("scan.snapshot-id", "1")); + TableDescriptor descriptor = TableDescriptorSerializer.from(travel, "db", "t"); + assertThat(descriptor.snapshotId()).isEqualTo(1L); + // The persisted schema must not carry the scan option. + assertThat(descriptor.tableSchema().options()).doesNotContainKey("scan.snapshot-id"); + } + + @Test + public void testNoSnapshotIdForOrdinaryScan() { + FileStoreTable table = createTable(); + writeOneRow(table); + TableDescriptor descriptor = TableDescriptorSerializer.from(table, "db", "t"); + assertThat(descriptor.snapshotId()).isNull(); + } + + private static void writeOneRow(FileStoreTable table) { + try (BatchTableWrite write = table.newBatchWriteBuilder().newWrite(); + BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + write.write(GenericRow.of(1, 1)); + commit.commit(write.prepareCommit()); + } catch (Exception e) { + throw new RuntimeException(e); + } + } +} diff --git a/paimon-core/src/test/resources/table/table-descriptor-v1.json b/paimon-core/src/test/resources/table/table-descriptor-v1.json new file mode 100644 index 000000000000..52134f204a2d --- /dev/null +++ b/paimon-core/src/test/resources/table/table-descriptor-v1.json @@ -0,0 +1,34 @@ +{ + "version" : 1, + "path" : "file:///tmp/warehouse/mydb.db/t", + "tableSchema" : { + "version" : 3, + "id" : 1, + "fields" : [ { + "id" : 0, + "name" : "id", + "type" : "INT NOT NULL" + }, { + "id" : 1, + "name" : "name", + "type" : "VARCHAR(100)" + }, { + "id" : 2, + "name" : "tags", + "type" : { + "type" : "ARRAY", + "element" : "INT" + } + } ], + "highestFieldId" : 2, + "partitionKeys" : [ ], + "primaryKeys" : [ "id" ], + "options" : { + "bucket" : "1" + }, + "comment" : "canonical", + "timeMillis" : 1700000000000 + }, + "database" : "mydb", + "name" : "t" +} \ No newline at end of file