Fix failure reading checkpoint created by CREATE OR REPLACE with different schema

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

@@ -1376,7 +1376,8 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
                 transactionLogWriter.flush();
 
                 if (replaceExistingTable) {
-                    writeCheckpointIfNeeded(session, schemaTableName, location, tableHandle.toCredentialsHandle(), tableHandle.getReadVersion(), checkpointInterval, commitVersion);
+                    verify(commitVersion > tableHandle.getReadVersion());
+                    writeCheckpointMandatory(session, schemaTableName, location, tableHandle.toCredentialsHandle(), commitVersion);
                 }
             }
         }
@@ -1748,7 +1749,7 @@ public Optional<ConnectorOutputMetadata> finishCreateTable(
             writeCommitted = true;
 
             if (handle.replace() && handle.readVersion().isPresent()) {
-                writeCheckpointIfNeeded(session, schemaTableName, handle.location(), handle.toCredentialsHandle(), handle.readVersion().getAsLong(), handle.checkpointInterval(), commitVersion);
+                writeCheckpointMandatory(session, schemaTableName, location, handle.toCredentialsHandle(), commitVersion);
             }
 
             if (isCollectExtendedStatisticsColumnStatisticsOnWrite(session) && !computedStatistics.isEmpty()) {
@@ -3182,9 +3183,7 @@ private void writeCheckpointIfNeeded(
             // This does not pose correctness issue but may be confusing if someone looks into transaction log.
             // To fix that we should allow for getting snapshot for given version.
 
-            TransactionLogReader transactionLogReader = new FileSystemTransactionLogReader(tableLocation, credentialsHandle, fileSystemFactory);
-            TableSnapshot snapshot = transactionLogAccess.loadSnapshot(session, transactionLogReader, table, tableLocation, Optional.of(newVersion), credentialsHandle);
-            checkpointWriterManager.writeCheckpoint(session, snapshot, credentialsHandle);
+            writeCheckpointMandatory(session, table, tableLocation, credentialsHandle, newVersion);
         }
         catch (Exception e) {
             // We can't fail here as transaction was already committed, in case of INSERT this could result
@@ -3193,6 +3192,14 @@ private void writeCheckpointIfNeeded(
         }
     }
 
+    private void writeCheckpointMandatory(ConnectorSession session, SchemaTableName table, String tableLocation, VendedCredentialsHandle credentialsHandle, long newVersion)
+            throws IOException
+    {
+        TransactionLogReader transactionLogReader = new FileSystemTransactionLogReader(tableLocation, credentialsHandle, fileSystemFactory);
+        TableSnapshot snapshot = transactionLogAccess.loadSnapshot(session, transactionLogReader, table, tableLocation, Optional.of(newVersion), credentialsHandle);
+        checkpointWriterManager.writeCheckpoint(session, snapshot, credentialsHandle);
+    }
+
     private void cleanupFailedWrite(ConnectorSession session, VendedCredentialsHandle credentialsHandle, List<DataFileInfo> dataFiles)
     {
         Location location = Location.of(credentialsHandle.tableLocation());

plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeConnectorTest.java

@@ -284,6 +284,42 @@ protected MaterializedResult getDescribeOrdersResult()
                 .build();
     }
 
+    @Test
+    void testCreateReplaceReadingCheckpointWithDifferentSchema()
+    {
+        try (TestTable table = newTrinoTable("test_create_replace_reading_checkpoint_", "(x int, y varchar) with (checkpoint_interval = 2)")) {
+            assertUpdate("INSERT INTO " + table.getName() + " VALUES (1, 'aa')", 1);
+            // generate a checkpoint
+            assertUpdate("INSERT INTO " + table.getName() + " VALUES (2, 'bb')", 1);
+
+            assertUpdate("CREATE OR REPLACE TABLE " + table.getName() + " (x int, y int) with (checkpoint_interval = 2)");
+            assertUpdate("INSERT INTO " + table.getName() + " VALUES (3, 3)", 1);
+            assertThat(query("SELECT * FROM " + table.getName()))
+                    .matches("VALUES (3, 3)");
+
+            assertUpdate("CREATE OR REPLACE TABLE " + table.getName() + " (z varchar)");
+            assertQueryReturnsEmptyResult("TABLE " + table.getName());
+        }
+    }
+
+    @Test
+    void testCreateReplaceReadingCheckpointWithDifferentSchemaCTAS()
+    {
+        try (TestTable table = newTrinoTable("test_create_replace_reading_checkpoint_", "(x int, y varchar) with (checkpoint_interval = 2)")) {
+            assertUpdate("INSERT INTO " + table.getName() + " VALUES (1, 'aa')", 1);
+            // generate a checkpoint
+            assertUpdate("INSERT INTO " + table.getName() + " VALUES (2, 'bb')", 1);
+
+            assertUpdate("CREATE OR REPLACE TABLE " + table.getName() + "  AS SELECT 3 AS x, 3 AS y", 1);
+            assertThat(query("SELECT * FROM " + table.getName()))
+                    .matches("VALUES (3, 3)");
+
+            assertUpdate("CREATE OR REPLACE TABLE " + table.getName() + " AS SELECT 'test' AS z", 1);
+            assertThat(query("TABLE " + table.getName()))
+                    .matches("VALUES VARCHAR 'test'");
+        }
+    }
+
     @Test
     @Override
     public void testShowCreateTable()