[Bug] Paimon Schema Creation Fails When Writing Directly to Paimon - "Cannot get latest schema for table" Error

### Search before asking

- [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar.

### Paimon version

Paimon 1.2.0

### Compute Engine

Flink 1.20.1
Paimon 1.2.0
Fluss 0.7.0

### Minimal reproduce step

### Problem
My task is to set up Paimon as a storage for Fluss (set up data synchronization from Fluss to Paimon)

When trying to set up data synchronization in Paimon and when trying to write directly to Paimon, I encounter the problem with inserts into the test table.

The exception thrown by the Paimon schema manager is "Cannot get latest schema for table test_paimon_direct".
The exception doesn't happen until I try to perform an SQL ```insert``` command.
Looks like for some reason Paimon skips the schema files creation upon the DDL statement execution, but then relies on the schema files when doing the commit.

Any advice would be highly appreciated.

More detail on the configuration and exception stack trace is below.

### Write directly to Paimon

I am running a cluster with Fluss (zookeeper, coordinator server, 3 tablet servers), flink-sql-client, flink-jobmanager, flink-taskmanager and flink-tiering-service.

To launch, I use .jar files from this site: https://fluss.apache.org/docs/maintenance/tiered-storage/lakehouse-storage/#enable-lakehouse-storage

Here is my current configuration:
```
version: '3.8'

services:
# Zookeeper
zookeeper:
image: zookeeper:3.9.2
restart: always
ports:
- "2181:2181"
networks:
- fluss-network

# Fluss Coordinator
coordinator-server:
image: fluss/fluss:0.7.0
command: coordinatorServer
depends_on:
- zookeeper
environment:
- |
FLUSS_PROPERTIES=
zookeeper.address: zookeeper:2181
bind.listeners: CLIENT://coordinator-server:9123,INTERNAL://coordinator-server:9124
advertised.listeners: CLIENT://coordinator-server:9123
internal.listener.name: INTERNAL
remote.data.dir: /tmp/fluss/remote-data
datalake.format: paimon
datalake.paimon.metastore: filesystem
datalake.paimon.warehouse: /tmp/paimon
volumes:
- shared-tmpfs:/tmp/paimon
ports:
- "9123:9123"
networks:
- fluss-network

# Fluss TabletServers
tablet-server-0:
image: fluss/fluss:0.7.0
command: tabletServer
depends_on:
- coordinator-server
environment:
- |
FLUSS_PROPERTIES=
zookeeper.address: zookeeper:2181
bind.listeners: INTERNAL://tablet-server-0:0,CLIENT://tablet-server-0:9123
advertised.listeners: CLIENT://tablet-server-0:9123
internal.listener.name: INTERNAL
tablet-server.id: 0
kv.snapshot.interval: 0s
data.dir: /tmp/fluss/data/tablet-server-0
remote.data.dir: /tmp/fluss/remote-data
datalake.format: paimon
datalake.paimon.metastore: filesystem
datalake.paimon.warehouse: /tmp/paimon
ports:
- "9124:9124"
volumes:
- shared-tmpfs:/tmp/paimon
networks:
- fluss-network

tablet-server-1:
image: fluss/fluss:0.7.0
command: tabletServer
depends_on:
- coordinator-server
environment:
- |
FLUSS_PROPERTIES=
zookeeper.address: zookeeper:2181
bind.listeners: INTERNAL://tablet-server-1:0,CLIENT://tablet-server-1:9125
advertised.listeners: CLIENT://tablet-server-1:9125
internal.listener.name: INTERNAL
tablet-server.id: 1
kv.snapshot.interval: 0s
data.dir: /tmp/fluss/data/tablet-server-1
remote.data.dir: /tmp/fluss/remote-data
datalake.format: paimon
datalake.paimon.metastore: filesystem
datalake.paimon.warehouse: /tmp/paimon
ports:
- "9125:9125"
volumes:
- shared-tmpfs:/tmp/paimon
networks:
- fluss-network

tablet-server-2:
image: fluss/fluss:0.7.0
command: tabletServer
depends_on:
- coordinator-server
environment:
- |
FLUSS_PROPERTIES=
zookeeper.address: zookeeper:2181
bind.listeners: INTERNAL://tablet-server-2:0,CLIENT://tablet-server-2:9126
advertised.listeners: CLIENT://tablet-server-2:9126
internal.listener.name: INTERNAL
tablet-server.id: 2
kv.snapshot.interval: 0s
data.dir: /tmp/fluss/data/tablet-server-2
remote.data.dir: /tmp/fluss/remote-data
datalake.format: paimon
datalake.paimon.metastore: filesystem
datalake.paimon.warehouse: /tmp/paimon
ports:
- "9126:9126"
volumes:
- shared-tmpfs:/tmp/paimon
networks:
- fluss-network

flink-jobmanager:
image: flink:1.20.1-scala_2.12
hostname: flink-jobmanager
container_name: flink-jobmanager
networks:
fluss-network:
aliases:
- flink-jobmanager
ports:
- "8081:8081"
- "6123:6123"
volumes:
- ./flink-conf.yaml:/opt/flink/conf/flink-conf.yaml
- ./fluss-flink-tiering-0.7.0.jar:/opt/flink/lib/fluss-flink-tiering-0.7.0.jar
- ./paimon-flink-1.20-1.2.0.jar:/opt/flink/lib/paimon-flink-1.20-1.2.0.jar
- ./flink-shaded-hadoop-2-uber-2.8.3-10.0.jar:/opt/flink/lib/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar
environment:
- JOB_MANAGER_RPC_ADDRESS=flink-jobmanager
- FLINK_PROPERTIES=jobmanager.rpc.address:flink-jobmanager
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8081"]
interval: 10s
timeout: 5s
retries: 10
command: ["jobmanager"]

flink-taskmanager:
image: flink:1.20.1-scala_2.12
hostname: flink-taskmanager
container_name: flink-taskmanager
depends_on:
- flink-jobmanager
networks:
fluss-network:
aliases:
- flink-taskmanager
ports:
- "6122:6122"
volumes:
- ./flink-conf.yaml:/opt/flink/conf/flink-conf.yaml
- ./flink-1.20.1/conf/log4j.properties:/opt/flink/conf/log4j.properties
- shared-tmpfs:/tmp/paimon
- ./data/paimon:/opt/flink/paimon
- ./fluss-flink-tiering-0.7.0.jar:/opt/flink/lib/fluss-flink-tiering-0.7.0.jar
- ./paimon-flink-1.20-1.2.0.jar:/opt/flink/lib/paimon-flink-1.20-1.2.0.jar
- ./flink-shaded-hadoop-2-uber-2.8.3-10.0.jar:/opt/flink/lib/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar
- ./flink-connector-files-1.20.1.jar:/opt/flink/lib/flink-connector-files-1.20.1.jar
- ./fluss-flink-1.20-0.7.0.jar:/opt/flink/lib/fluss-flink-1.20-0.7.0.jar
command: taskmanager
entrypoint: ["sh", "-c", "apt-get update && apt-get install -y netcat-openbsd && apt-get update && apt-get install -y net-tools iputils-ping && /docker-entrypoint.sh taskmanager"]
environment:
- JOB_MANAGER_RPC_ADDRESS=flink-jobmanager
- FLINK_PROPERTIES=jobmanager.rpc.address:flink-jobmanager

# Flink SQL Client

flink-sql-client:
image: flink:1.20.1-scala_2.12
container_name: flink-sql-client
depends_on:
- coordinator-server
- flink-jobmanager
networks:
- fluss-network
volumes:
- ./flink-conf.yaml:/opt/flink/conf/flink-conf.yaml
- ./paimon-flink-1.20-1.2.0.jar:/opt/flink/lib/paimon-flink-1.20-1.2.0.jar
- ./flink-shaded-hadoop-2-uber-2.8.3-10.0.jar:/opt/flink/lib/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar
- ./flink-connector-files-1.20.1.jar:/opt/flink/lib/flink-connector-files-1.20.1.jar
- ./fluss-flink-1.20-0.7.0.jar:/opt/flink/lib/fluss-flink-1.20-0.7.0.jar
- ./fluss-flink-tiering-0.7.0.jar:/opt/flink/lib/fluss-flink-tiering-0.7.0.jar
- ./init.sql:/opt/flink/init.sql
- shared-tmpfs:/tmp/paimon
environment:
- FLINK_HOME=/opt/flink
- HADOOP_CLASSPATH=/opt/flink/lib/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar
tty: true
stdin_open: true
entrypoint: ["/bin/bash", "-c", "apt-get update && apt-get install -y netcat && while ! nc -z coordinator-server 9123; do sleep 2; done; while ! nc -z flink-jobmanager 6123; do sleep 2; done; /opt/flink/bin/sql-client.sh -f /opt/flink/init.sql; while true; do sleep 3600; done"]

flink-tiering-service:
image: flink:1.20.1-scala_2.12
container_name: flink-tiering-service
depends_on:
flink-jobmanager:
condition: service_healthy
coordinator-server:
condition: service_started
networks:
fluss-network:
aliases:
- flink-tiering-service
volumes:
- ./fluss-flink-tiering-0.7.0.jar:/opt/flink/lib/fluss-flink-tiering-0.7.0.jar
- ./paimon-flink-1.20-1.2.0.jar:/opt/flink/lib/paimon-flink-1.20-1.2.0.jar
- ./flink-shaded-hadoop-2-uber-2.8.3-10.0.jar:/opt/flink/lib/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar
- ./flink-connector-files-1.20.1.jar:/opt/flink/lib/flink-connector-files-1.20.1.jar
- ./fluss-flink-1.20-0.7.0.jar:/opt/flink/lib/fluss-flink-1.20-0.7.0.jar
- ./fluss-server-0.7.0.jar:/opt/flink/lib/fluss-server.jar
- ./fluss-lake-paimon.jar:/opt/flink/lib/fluss-lake-paimon.jar
- ./flink-conf.yaml:/opt/flink/conf/flink-conf.yaml
- shared-tmpfs:/tmp/paimon
environment:
FLINK_PROPERTIES: |
jobmanager.rpc.address: flink-jobmanager
rest.address: flink-jobmanager
rest.port: 8081
rest.connection-timeout: 30000
rest.retry.delay: 10000
lake.storage.plugins.dir: file:///opt/flink/lib
plugin.lake-storage-dir: file:///opt/flink/lib
classloader.resolve-order: parent-first
classloader.parent-first-patterns-add: com.alibaba.fluss*
flink.version: 1.20.1
fluss.version: 0.7.0
paimon.version: 1.2.0
FLINK_JOBMANAGER_ADDRESS: flink-jobmanager
FLINK_JOBMANAGER_PORT: 6123
FLINK_REST_ADDRESS: flink-jobmanager
FLINK_REST_PORT: 8081
entrypoint: ["/opt/flink/bin/taskmanager.sh", "start-foreground"]
restart: unless-stopped

volumes:
shared-tmpfs:
driver: local
driver_opts:
type: "tmpfs"
device: "tmpfs"
o: rw,uid=9999,gid=9999
grafana-storage:
driver: local

networks:
fluss-network:
driver: bridge
attachable: true
internal: false

```

I am running the following sql queries:
```
SET sql-client.execution.result-mode=TABLEAU;

SET 'paimon.schema.auto-update' = 'true';
SET 'paimon.merge-engine' = 'deduplicate';

CREATE CATALOG paimon WITH (
'type' = 'paimon',
'warehouse' = 'file:///opt/flink/paimon'
);

USE CATALOG paimon;

CREATE DATABASE IF NOT EXISTS my_database;

USE my_database;

DROP TABLE IF EXISTS test_paimon_direct;

CREATE TABLE IF NOT EXISTS test_paimon_direct (
Client_ID INT PRIMARY KEY NOT ENFORCED,
Surname STRING,
Name STRING,
Account_number BIGINT,
Time_Stamp BIGINT
) WITH (
'bucket' = '1',
'merge-engine' = 'deduplicate',
'snapshot.time-retained' = '1h',
'write-buffer-spillable' = 'false',
'manifest-format' = 'avro',
'changelog-producer' = 'none',
'fs.allowed-fallback-filesystems' = 'file'
);

DESCRIBE test_paimon_direct;

SELECT * FROM test_paimon_direct;
```

When running this way, I don't have a table schema created (but something about the table is saved, some information about it) :
```
$ docker-compose exec flink-taskmanager ls -la /opt/flink/paimon/my_database.db/test_paimon_direct/
WARN[0000] /home/akulishova/docker_compose_cluster/docker-compose.yml: the attribute `version` is obsolete, it will be ignored, please remove it to avoid potential confusion
total 16
drwxr-sr-x 4 flink 1008 4096 Jul 28 03:54 .
drwxr-sr-x 3 flink 1008 4096 Jul 28 03:54 ..
drwxr-sr-x 2 flink 1008 4096 Jul 28 05:17 bucket-0
drwxr-sr-x 2 flink 1008 4096 Jul 28 05:20 manifest
```

And I get this error in the taskmanager logs:
```
2025-07-28 04:46:51,355 WARN org.apache.flink.runtime.taskmanager.Task [] - Writer : test_paimon_direct -> Global Committer : test_paimon_direct -> end: Writer (1/1)#6 (de67d09f13a0b5f54bdab904cfdd7cc0_f6dc7f4d2283f4605b127b9364e21148_0_6) switched from RUNNING to FAILED with failure cause:
java.lang.RuntimeException: Exception occurs when preparing snapshot #1 by user 9ef0554f-cce4-41f9-a5f9-f432d35b3e1d with hash 9223372036854775807 and kind APPEND. Clean up.
at org.apache.paimon.operation.FileStoreCommitImpl.tryCommitOnce(FileStoreCommitImpl.java:1081) ~[paimon-flink-1.20-1.2.0.jar:1.2.0]
at org.apache.paimon.operation.FileStoreCommitImpl.tryCommit(FileStoreCommitImpl.java:782) ~[paimon-flink-1.20-1.2.0.jar:1.2.0]
at org.apache.paimon.operation.FileStoreCommitImpl.commit(FileStoreCommitImpl.java:325) ~[paimon-flink-1.20-1.2.0.jar:1.2.0]
at org.apache.paimon.table.sink.TableCommitImpl.commitMultiple(TableCommitImpl.java:217) ~[paimon-flink-1.20-1.2.0.jar:1.2.0]
at org.apache.paimon.table.sink.TableCommitImpl.filterAndCommitMultiple(TableCommitImpl.java:256) ~[paimon-flink-1.20-1.2.0.jar:1.2.0]
at org.apache.paimon.flink.sink.StoreCommitter.filterAndCommit(StoreCommitter.java:121) ~[paimon-flink-1.20-1.2.0.jar:1.2.0]
at org.apache.paimon.flink.sink.CommitterOperator.commitUpToCheckpoint(CommitterOperator.java:213) ~[paimon-flink-1.20-1.2.0.jar:1.2.0]
at org.apache.paimon.flink.sink.CommitterOperator.notifyCheckpointComplete(CommitterOperator.java:192) ~[paimon-flink-1.20-1.2.0.jar:1.2.0]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.notifyCheckpointComplete(StreamOperatorWrapper.java:104) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.notifyCheckpointComplete(RegularOperatorChain.java:145) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.notifyCheckpoint(SubtaskCheckpointCoordinatorImpl.java:478) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.notifyCheckpointComplete(SubtaskCheckpointCoordinatorImpl.java:411) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.StreamTask.notifyCheckpointComplete(StreamTask.java:1565) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$notifyCheckpointCompleteAsync$20(StreamTask.java:1506) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$notifyCheckpointOperation$23(StreamTask.java:1545) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:101) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:414) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:383) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:368) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:973) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:917) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:970) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:949) [flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:763) [flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575) [flink-dist-1.20.1.jar:1.20.1]
at java.lang.Thread.run(Unknown Source) [?:?]
Caused by: java.lang.RuntimeException: Cannot get latest schema for table test_paimon_direct
at org.apache.paimon.schema.SchemaManager.lambda$latestOrThrow$0(SchemaManager.java:124) ~[paimon-flink-1.20-1.2.0.jar:1.2.0]
at java.util.Optional.orElseThrow(Unknown Source) ~[?:?]
at org.apache.paimon.schema.SchemaManager.latestOrThrow(SchemaManager.java:124) ~[paimon-flink-1.20-1.2.0.jar:1.2.0]
at org.apache.paimon.operation.FileStoreCommitImpl.tryCommitOnce(FileStoreCommitImpl.java:1033) ~[paimon-flink-1.20-1.2.0.jar:1.2.0]
... 27 more

```

The contents of the flink-conf file.yaml:
```
$ cat flink-conf.yaml
metrics.reporter.prom.interval: 10 SECONDS
blob.server.port: 6124
taskmanager.memory.process.size: 3072m
metrics.reporter.prom.port: 9250-9260
jobmanager.rpc.address: flink-jobmanager
metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory
jobmanager.memory.process.size: 2048m
jobmanager.rpc.port: 6123
taskmanager.rpc.port: 6122
query.server.port: 6125
rest.bind-address: 0.0.0.0
logger.paimon.name: org.apache.paimon
rest.port: 8081
execution.checkpointing.interval: 1 min
jobmanager.bind-host: 0.0.0.0
table.exec.state.ttl: 3 h
logger.paimon.level: TRACE
taskmanager.host: flink-taskmanager
taskmanager.memory.managed.size: 512m
logger.checkpoint.level: DEBUG
logger.checkpoint.name: org.apache.flink.runtime.checkpoint
taskmanager.numberOfTaskSlots: 2
taskmanager.memory.task.heap.size: 1024m
metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter
```

At the same time, when executing sql commands, I get the following output (that is, the DESCRIBE command works and takes information about the table somewhere):
```
Command history file path: /root/.flink-sql-history
Flink SQL> [INFO] Execute statement succeeded.

Flink SQL>
> -- =============================================
> -- 1. Создаём каталог Paimon (если ещё не создан)[INFO] Execute statement succeeded.

Flink SQL> [INFO] Execute statement succeeded.

Flink SQL>
> CREATE CATALOG paimon WITH (
> 'type' = 'paimon',
> 'warehouse' = 'file:///opt/flink/paimon'
> )[INFO] Execute statement succeeded.

Flink SQL> [INFO] Execute statement succeeded.

Flink SQL> [INFO] Execute statement succeeded.

Flink SQL> [INFO] Execute statement succeeded.

Flink SQL> [INFO] Execute statement succeeded.

Flink SQL>
> CREATE TABLE IF NOT EXISTS test_paimon_direct (
> Client_ID INT PRIMARY KEY NOT ENFORCED,
> Surname STRING,
> Name STRING,
> Account_number BIGINT,
> Time_Stamp BIGINT
> ) WITH (
> 'bucket' = '1',
> 'merge-engine' = 'deduplicate',
> 'snapshot.time-retained' = '1h',
> 'write-buffer-spillable' = 'false', -- Отключаем spill на диск
> 'manifest-format' = 'avro',
> 'changelog-producer' = 'none',
> 'fs.allowed-fallback-filesystems' = 'file' -- Явно разрешаем локальную FS
> )[INFO] Execute statement succeeded.

Flink SQL> +----------------+--------+-------+----------------+--------+-----------+
| name | type | null | key | extras | watermark |
+----------------+--------+-------+----------------+--------+-----------+
| Client_ID | INT | FALSE | PRI(Client_ID) | | |
| Surname | STRING | TRUE | | | |
| Name | STRING | TRUE | | | |
| Account_number | BIGINT | TRUE | | | |
| Time_Stamp | BIGINT | TRUE | | | |
+----------------+--------+-------+----------------+--------+-----------+
5 rows in set

Flink SQL> +----+-------------+--------------------------------+--------------------------------+----------------------+----------------------+
| op | Client_ID | Surname | Name | Account_number | Time_Stamp |
+----+-------------+--------------------------------+--------------------------------+----------------------+----------------------+

```

### Setting up synchronization in Paimon when writing to Fluss

When trying to set up synchronization (actually, this is my main task, an attempt to write directly to paimon was made to find hints to solve the problem), the same cluster configuration was used.

The following sql commands were used:
```
$ cat init.sql

SET sql-client.execution.result-mode=TABLEAU;

SET 'execution.checkpointing.interval' = '10s';

CREATE CATALOG fluss WITH (
'type' = 'fluss',
'bootstrap.servers' = 'coordinator-server:9123'
);

SHOW CATALOGS;

USE CATALOG fluss;

CREATE DATABASE IF NOT EXISTS my_database;

USE my_database;

CREATE TABLE IF NOT EXISTS my_table (
Client_ID INT,
Surname STRING,
Name STRING,
Account_number BIGINT,
First_transaction_day_time STRING,
Last_transaction_day_time STRING,
First_transaction_sum STRING,
Last_transaction_sum STRING,
Day_transaction_count STRING,
Day_transaction_sum STRING,
Day_transaction_avg_sum STRING,
Account_number_Surname STRING,
Account_number_Name STRING,
Name_surname_day_transaction_sum STRING,
Name_surname_day_transaction_count STRING,
Name_surname_day_transaction_avg_sum STRING,
Name_surname_day_transaction_sum_account_number STRING,
Name_surname_day_transaction_avg_sum_account_number STRING,
Time_Stamp BIGINT,
PRIMARY KEY (`Client_ID`) NOT ENFORCED
) WITH (
'connector' = 'fluss',
'bootstrap.servers' = 'coordinator-server:9123',
'table.datalake.enabled' = 'true',
'table.datalake.freshness' = '30s'
);

SHOW CURRENT CATALOG;

SHOW CURRENT DATABASE;

SHOW TABLES;

DESCRIBE my_table;

INSERT INTO my_table VALUES
(1, 'Ivanov', 'Ivan', 123456789, '2023-01-01 10:00:00', '2023-01-01 18:00:00', '100.50', '500.75', '5', '1500.25', '300.05', '123456789_Ivanov', '123456789_Ivan', '1500.25_Ivanov_Ivan', '5_Ivanov_Ivan', '300.05_Ivanov_Ivan', '1500.25_123456789', '300.05_123456789', 1672531200000),
(2, 'Petrov', 'Petr', 987654321, '2023-01-02 09:30:00', '2023-01-02 17:45:00', '200.00', '750.30', '3', '1450.90', '483.63', '987654321_Petrov', '987654321_Petr', '1450.90_Petrov_Petr', '3_Petrov_Petr', '483.63_Petrov_Petr', '1450.90_987654321', '483.63_987654321', 1672617600000);

```

The resulting output:
```
Command history file path: /root/.flink-sql-history
Flink SQL> [INFO] Execute statement succeeded.

Flink SQL> [INFO] Execute statement succeeded.

Flink SQL>
> CREATE CATALOG fluss WITH (
> 'type' = 'fluss',
> 'bootstrap.servers' = 'coordinator-server:9123'
> )[INFO] Execute statement succeeded.

Flink SQL> +-----------------+
| catalog name |
+-----------------+
| default_catalog |
| fluss |
+-----------------+
2 rows in set

Flink SQL> [INFO] Execute statement succeeded.

Flink SQL> [INFO] Execute statement succeeded.

Flink SQL> [INFO] Execute statement succeeded.

Flink SQL>
> CREATE TABLE IF NOT EXISTS my_table (
> Client_ID INT,
> Surname STRING,
> Name STRING,
> Account_number BIGINT,
> First_transaction_day_time STRING,
> Last_transaction_day_time STRING,
> First_transaction_sum STRING,
> Last_transaction_sum STRING,
> Day_transaction_count STRING,
> Day_transaction_sum STRING,
> Day_transaction_avg_sum STRING,
> Account_number_Surname STRING,
> Account_number_Name STRING,
> Name_surname_day_transaction_sum STRING,
> Name_surname_day_transaction_count STRING,
> Name_surname_day_transaction_avg_sum STRING,
> Name_surname_day_transaction_sum_account_number STRING,
> Name_surname_day_transaction_avg_sum_account_number STRING,
> Time_Stamp BIGINT,
> PRIMARY KEY (`Client_ID`) NOT ENFORCED
> ) WITH (
> 'connector' = 'fluss',
> 'bootstrap.servers' = 'coordinator-server:9123',
> 'table.datalake.enabled' = 'true',
> 'table.datalake.freshness' = '30s'
> )[INFO] Execute statement succeeded.

Flink SQL> +----------------------+
| current catalog name |
+----------------------+
| fluss |
+----------------------+
1 row in set

Flink SQL> +-----------------------+
| current database name |
+-----------------------+
| my_database |
+-----------------------+
1 row in set

Flink SQL> +------------+
| table name |
+------------+
| my_table |
+------------+
1 row in set

Flink SQL> +-----------------------------------------------------+--------+-------+----------------+--------+-----------+
| name | type | null | key | extras | watermark |
+-----------------------------------------------------+--------+-------+----------------+--------+-----------+
| Client_ID | INT | FALSE | PRI(Client_ID) | | |
| Surname | STRING | TRUE | | | |
| Name | STRING | TRUE | | | |
| Account_number | BIGINT | TRUE | | | |
| First_transaction_day_time | STRING | TRUE | | | |
| Last_transaction_day_time | STRING | TRUE | | | |
| First_transaction_sum | STRING | TRUE | | | |
| Last_transaction_sum | STRING | TRUE | | | |
| Day_transaction_count | STRING | TRUE | | | |
| Day_transaction_sum | STRING | TRUE | | | |
| Day_transaction_avg_sum | STRING | TRUE | | | |
| Account_number_Surname | STRING | TRUE | | | |
| Account_number_Name | STRING | TRUE | | | |
| Name_surname_day_transaction_sum | STRING | TRUE | | | |
| Name_surname_day_transaction_count | STRING | TRUE | | | |
| Name_surname_day_transaction_avg_sum | STRING | TRUE | | | |
| Name_surname_day_transaction_sum_account_number | STRING | TRUE | | | |
| Name_surname_day_transaction_avg_sum_account_number | STRING | TRUE | | | |
| Time_Stamp | BIGINT | TRUE | | | |
+-----------------------------------------------------+--------+-------+----------------+--------+-----------+
19 rows in set

Flink SQL>
>
> INSERT INTO my_table VALUES
> (1, 'Ivanov', 'Ivan', 123456789, '2023-01-01 10:00:00', '2023-01-01 18:00:00', '100.50', '500.75', '5', '1500.25', '300.05', '123456789_Ivanov', '123456789_Ivan', '1500.25_Ivanov_Ivan', '5_Ivanov_Ivan', '300.05_Ivanov_Ivan', '1500.25_123456789', '300.05_123456789', 1672531200000),
> (2, 'Petrov', 'Petr', 987654321, '2023-01-02 09:30:00', '2023-01-02 17:45:00', '200.00', '750.30', '3', '1450.90', '483.63', '987654321_Petrov', '987654321_Petr', '1450.90_Petrov_Petr', '3_Petrov_Petr', '483.63_Petrov_Petr', '1450.90_987654321', '483.63_987654321', 1672617600000)[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 72269a47c6091fdd65afa8dc85123935
```

In this case, the table schema is saved and I can see it inside the container.
```
$ docker-compose exec tablet-server-1 ls -la /tmp/paimon/my_database.db/my_table/schema
WARN[0000] /home/akulishova/docker_compose_cluster/docker-compose.yml: the attribute `version` is obsolete, it will be ignored, please remove it to avoid potential confusion
total 4
drwxr-xr-x 2 root root 60 Jul 28 07:37 .
drwxr-xr-x 3 root root 60 Jul 28 07:37 ..
-rw-r--r-- 1 root root 2375 Jul 28 07:37 schema-0
```

I suspect that the main problem of data synchronization is related to incorrect execution of checkpoints. Here are my logs:
```
docker logs flink-jobmanager | grep "checkpoint"
INFO [] - Loading configuration property: logger.checkpoint.level, DEBUG
INFO [] - Loading configuration property: logger.checkpoint.name, org.apache.flink.runtime.checkpoint
INFO [] - Loading configuration property: execution.checkpointing.interval, 1 min
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.hadoop.security.authentication.util.KerberosUtil (file:/opt/flink/lib/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar) to method sun.security.krb5.Config.getInstance()
WARNING: Please consider reporting this to the maintainers of org.apache.hadoop.security.authentication.util.KerberosUtil
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
2025-07-28 07:37:08,638 INFO org.apache.flink.configuration.GlobalConfiguration [] - Loading configuration property: logger.checkpoint.level, DEBUG
2025-07-28 07:37:08,638 INFO org.apache.flink.configuration.GlobalConfiguration [] - Loading configuration property: logger.checkpoint.name, org.apache.flink.runtime.checkpoint
2025-07-28 07:37:08,639 INFO org.apache.flink.configuration.GlobalConfiguration [] - Loading configuration property: execution.checkpointing.interval, 1 min
2025-07-28 07:37:24,853 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - No checkpoint found during restore.
2025-07-28 07:37:26,608 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Triggering a manual checkpoint for job 72269a47c6091fdd65afa8dc85123935.
2025-07-28 07:37:26,625 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Triggering checkpoint 1 (type=CheckpointType{name='Checkpoint', sharingFilesStrategy=FORWARD_BACKWARD}) @ 1753688246613 for job 72269a47c6091fdd65afa8dc85123935.
2025-07-28 07:37:26,712 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Completed checkpoint 1 for job 72269a47c6091fdd65afa8dc85123935 (0 bytes, checkpointDuration=93 ms, finalizationTime=5 ms).
2025-07-28 07:37:29,562 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Triggering checkpoint 2 (type=CheckpointType{name='Checkpoint', sharingFilesStrategy=FORWARD_BACKWARD}) @ 1753688249559 for job 72269a47c6091fdd65afa8dc85123935.
2025-07-28 07:37:29,616 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Triggering Checkpoint 2 for job 72269a47c6091fdd65afa8dc85123935 failed due to org.apache.flink.runtime.rpc.exceptions.RecipientUnreachableException: Could not send message [RemoteRpcInvocation(TaskExecutorGateway.triggerCheckpoint(ExecutionAttemptID, long, long, CheckpointOptions))] from sender [Actor[pekko.tcp://flink@flink-jobmanager:6123/temp/taskmanager_0$x]] to recipient [Actor[pekko://flink/user/rpc/taskmanager_0#-2000800266]], because the recipient is unreachable. This can either mean that the recipient has been terminated or that the remote RpcService is currently not reachable.
2025-07-28 07:37:29,625 WARN org.apache.flink.runtime.checkpoint.CheckpointFailureManager [] - Failed to trigger or complete checkpoint 2 for job 72269a47c6091fdd65afa8dc85123935. (0 consecutive failed attempts so far)
org.apache.flink.runtime.checkpoint.CheckpointException: Trigger checkpoint failure.
at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.lambda$triggerCheckpointRequest$10(CheckpointCoordinator.java:810) ~[flink-dist-1.20.1.jar:1.20.1]
2025-07-28 07:37:39,561 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Triggering checkpoint 3 (type=CheckpointType{name='Checkpoint', sharingFilesStrategy=FORWARD_BACKWARD}) @ 1753688259558 for job 72269a47c6091fdd65afa8dc85123935.
2025-07-28 07:37:39,579 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Triggering Checkpoint 3 for job 72269a47c6091fdd65afa8dc85123935 failed due to org.apache.flink.runtime.rpc.exceptions.RecipientUnreachableException: Could not send message [RemoteRpcInvocation(TaskExecutorGateway.triggerCheckpoint(ExecutionAttemptID, long, long, CheckpointOptions))] from sender [Actor[pekko.tcp://flink@flink-jobmanager:6123/temp/taskmanager_0$E]] to recipient [Actor[pekko://flink/user/rpc/taskmanager_0#-2000800266]], because the recipient is unreachable. This can either mean that the recipient has been terminated or that the remote RpcService is currently not reachable.
2025-07-28 07:37:39,582 WARN org.apache.flink.runtime.checkpoint.CheckpointFailureManager [] - Failed to trigger or complete checkpoint 3 for job 72269a47c6091fdd65afa8dc85123935. (0 consecutive failed attempts so far)
org.apache.flink.runtime.checkpoint.CheckpointException: Trigger checkpoint failure.
at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.lambda$triggerCheckpointRequest$10(CheckpointCoordinator.java:810) ~[flink-dist-1.20.1.jar:1.20.1]
2025-07-28 07:37:42,120 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Restoring job 72269a47c6091fdd65afa8dc85123935 from Checkpoint 1 @ 1753688246613 for 72269a47c6091fdd65afa8dc85123935 located at .
2025-07-28 07:37:42,126 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - No master state to restore
2025-07-28 07:37:43,350 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Triggering a manual checkpoint for job 72269a47c6091fdd65afa8dc85123935.
2025-07-28 07:37:43,352 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Triggering checkpoint 4 (type=CheckpointType{name='Checkpoint', sharingFilesStrategy=FORWARD_BACKWARD}) @ 1753688263351 for job 72269a47c6091fdd65afa8dc85123935.
2025-07-28 07:37:43,395 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Completed checkpoint 4 for job 72269a47c6091fdd65afa8dc85123935 (0 bytes, checkpointDuration=35 ms, finalizationTime=9 ms).
2025-07-28 07:37:44,511 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Triggering checkpoint 5 (type=CheckpointType{name='Checkpoint', sharingFilesStrategy=FORWARD_BACKWARD}) @ 1753688264508 for job 72269a47c6091fdd65afa8dc85123935.
2025-07-28 07:37:44,543 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Decline checkpoint 5 by task bfbe6f0cc45bee031b658e30fbf92934_306d8342cb5b2ad8b53f1be57f65bee8_0_1 of job 72269a47c6091fdd65afa8dc85123935 at flink-taskmanager:6122-6f0ae1 @ flink-taskmanager.docker_compose_cluster_fluss-network (dataPort=34137).
org.apache.flink.util.SerializedThrowable: org.apache.flink.runtime.checkpoint.CheckpointException: Task name with subtask : my_table[3]: Writer (1/1)#1 Failure reason: Checkpoint was declined (task is closing)
2025-07-28 07:37:44,557 WARN org.apache.flink.runtime.checkpoint.CheckpointFailureManager [] - Failed to trigger or complete checkpoint 5 for job 72269a47c6091fdd65afa8dc85123935. (0 consecutive failed attempts so far)
org.apache.flink.runtime.checkpoint.CheckpointException: Checkpoint was declined (task is closing)
at org.apache.flink.runtime.messages.checkpoint.SerializedCheckpointException.unwrap(SerializedCheckpointException.java:51) ~[flink-dist-1.20.1.jar:1.20.1]
at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.receiveDeclineMessage(CheckpointCoordinator.java:1145) ~[flink-dist-1.20.1.jar:1.20.1]
Caused by: org.apache.flink.util.SerializedThrowable: org.apache.flink.runtime.checkpoint.CheckpointException: Task name with subtask : my_table[3]: Writer (1/1)#1 Failure reason: Checkpoint was declined (task is closing)
2025-07-28 07:37:45,570 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Stopping checkpoint coordinator for job 72269a47c6091fdd65afa8dc85123935.
2025-07-28 07:37:45,615 INFO org.apache.flink.runtime.checkpoint.StandaloneCompletedCheckpointStore [] - Shutting down
```

To be honest, at the moment, I just don't know what I need to wear and what settings I need to fix.

### What doesn't meet your expectations?

I suspect that the main problem of data synchronization is related to incorrect execution of checkpoints.

### Anything else?

_No response_

### Are you willing to submit a PR?

- [ ] I'm willing to submit a PR!

Start with the Docker Compose configuration, init.sql, and the taskmanager logs, then reproduce the CREATE TABLE and INSERT sequence described in the report. Compare the table directory contents before and after the insert and verify whether the schema files are created and the insert completes without the "Cannot get latest schema for table" error.

Written by the indexing model from the issue text.