@@ -337,6 +337,154 @@ private void testAddNestedColumnWithColumnMappingMode(String columnMappingMode)
337337 .containsPattern ("(delta\\ .columnMapping\\ .physicalName.*?){11}" );
338338 }
339339
340+ /**
341+ * @see databricks164.test_write_checkpoint_on_schema_change
342+ */
343+ @ Test
344+ void testDatabricksWriteCheckpointOnSchemaChange ()
345+ throws Exception
346+ {
347+ String tableName = "test_dbx_write_checkpoint_on_schema_change" + randomNameSuffix ();
348+ Path tableLocation = catalogDir .resolve (tableName );
349+ copyDirectoryContents (new File (Resources .getResource ("databricks164/test_write_checkpoint_on_schema_change" ).toURI ()).toPath (), tableLocation );
350+ assertUpdate ("CALL system.register_table(CURRENT_SCHEMA, '%s', '%s')" .formatted (tableName , tableLocation .toUri ()));
351+
352+ Path tableLocationPath = Path .of (getTableLocation (tableName ).replace ("file://" , "" ));
353+
354+ // verify checkpoint at version 3 not exists
355+ // alter table to add a new column not treated as schema change, thus no new checkpoint should be created
356+ assertThat (Files .exists (tableLocationPath .resolve ("_delta_log/00000000000000000003.checkpoint.parquet" ))).isFalse ();
357+ assertThat (query ("SELECT * FROM " + tableName + " FOR VERSION AS OF 4" ))
358+ .matches ("VALUES (4, varchar 'version4')" );
359+
360+ // verify checkpoint at version 5 exists
361+ // alter table to drop a column treated as schema change, thus new checkpoint should be created
362+ assertThat (Files .exists (tableLocationPath .resolve ("_delta_log/00000000000000000005.checkpoint.parquet" ))).isTrue ();
363+ assertThat (query ("SELECT * FROM " + tableName + " FOR VERSION AS OF 6" ))
364+ .matches ("VALUES varchar 'version6'" );
365+
366+ // verify checkpoint at version 7 exists
367+ // alter table to rename a column treated as schema change, thus new checkpoint should be created
368+ assertThat (Files .exists (tableLocationPath .resolve ("_delta_log/00000000000000000007.checkpoint.parquet" ))).isTrue ();
369+ assertThat (query ("SELECT * FROM " + tableName + " FOR VERSION AS OF 8" ))
370+ .matches ("VALUES varchar 'version8'" );
371+
372+ // verify checkpoint at version 9 exists
373+ // alter table to change column type treated as schema change, thus new checkpoint should be created
374+ assertThat (Files .exists (tableLocationPath .resolve ("_delta_log/00000000000000000009.checkpoint.parquet" ))).isTrue ();
375+ assertThat (query ("SELECT * FROM " + tableName + " FOR VERSION AS OF 10" ))
376+ .matches ("VALUES 10" );
377+
378+ // verify checkpoint at version 11 exists
379+ // alter table to change column type back treated as schema change, thus new checkpoint should be created
380+ assertThat (Files .exists (tableLocationPath .resolve ("_delta_log/00000000000000000011.checkpoint.parquet" ))).isTrue ();
381+ assertThat (query ("SELECT * FROM " + tableName + " FOR VERSION AS OF 12" ))
382+ .matches ("VALUES varchar 'version12'" );
383+ }
384+
385+ @ Test
386+ void testWriteCheckpointOnSchemaChange ()
387+ {
388+ try (TestTable table = newTrinoTable ("test_write_checkpoint_on_schema_change" , "(x int) WITH (checkpoint_interval = 2)" )) {
389+ Path tableLocation = Path .of (getTableLocation (table .getName ()).replace ("file://" , "" ));
390+
391+ assertUpdate ("INSERT INTO " + table .getName () + " VALUES 1" , 1 );
392+ // generate checkpoint at version 2
393+ assertUpdate ("INSERT INTO " + table .getName () + " VALUES 2" , 1 );
394+ assertThat (query ("TABLE " + table .getName ()))
395+ .matches ("VALUES 1, 2" );
396+ assertThat (Files .exists (tableLocation .resolve ("_delta_log/00000000000000000002.checkpoint.parquet" ))).isTrue ();
397+
398+ // alter table to add a new column at version 3
399+ assertUpdate ("CREATE OR REPLACE TABLE " + table .getName () + " (x int, y varchar) WITH (checkpoint_interval = 20)" );
400+ // alter table to add a new column not treated as schema change, thus no new checkpoint should be created
401+ assertThat (Files .exists (tableLocation .resolve ("_delta_log/00000000000000000003.checkpoint.parquet" ))).isFalse ();
402+ assertUpdate ("INSERT INTO " + table .getName () + " VALUES (4, 'version4')" , 1 );
403+ assertThat (query ("TABLE " + table .getName ()))
404+ .matches ("VALUES (4, varchar 'version4')" );
405+
406+ // alter table to drop a column at version 5
407+ assertUpdate ("CREATE OR REPLACE TABLE " + table .getName () + " (y varchar) WITH (checkpoint_interval = 20)" );
408+ // alter table to drop a column treated as schema change, thus new checkpoint should be created
409+ assertThat (Files .exists (tableLocation .resolve ("_delta_log/00000000000000000005.checkpoint.parquet" ))).isTrue ();
410+ assertUpdate ("INSERT INTO " + table .getName () + " VALUES 'version6'" , 1 );
411+ assertThat (query ("TABLE " + table .getName ()))
412+ .matches ("VALUES varchar 'version6'" );
413+
414+ // alter table to rename a column at version 7
415+ assertUpdate ("CREATE OR REPLACE TABLE " + table .getName () + " (z varchar) WITH (checkpoint_interval = 20)" );
416+ // alter table to rename a column treated as schema change, thus new checkpoint should be created
417+ assertThat (Files .exists (tableLocation .resolve ("_delta_log/00000000000000000007.checkpoint.parquet" ))).isTrue ();
418+ assertUpdate ("INSERT INTO " + table .getName () + " VALUES 'version8'" , 1 );
419+ assertThat (query ("TABLE " + table .getName ()))
420+ .matches ("VALUES varchar 'version8'" );
421+
422+ // alter table to change column type at version 9
423+ assertUpdate ("CREATE OR REPLACE TABLE " + table .getName () + " (z int) WITH (checkpoint_interval = 20)" );
424+ // alter table to change column type treated as schema change, thus new checkpoint should be created
425+ assertThat (Files .exists (tableLocation .resolve ("_delta_log/00000000000000000009.checkpoint.parquet" ))).isTrue ();
426+ assertUpdate ("INSERT INTO " + table .getName () + " VALUES 10" , 1 );
427+ assertThat (query ("TABLE " + table .getName ()))
428+ .matches ("VALUES 10" );
429+
430+ // alter table to change column type back at version 11, test change type again
431+ assertUpdate ("CREATE OR REPLACE TABLE " + table .getName () + " (z varchar) WITH (checkpoint_interval = 20)" );
432+ assertThat (Files .exists (tableLocation .resolve ("_delta_log/00000000000000000011.checkpoint.parquet" ))).isTrue ();
433+ assertUpdate ("INSERT INTO " + table .getName () + " VALUES 'version12'" , 1 );
434+ assertThat (query ("TABLE " + table .getName ()))
435+ .matches ("VALUES varchar 'version12'" );
436+ }
437+ }
438+
439+ @ Test
440+ void testWriteCheckpointOnSchemaChangeCTAS ()
441+ {
442+ try (TestTable table = newTrinoTable ("test_write_checkpoint_on_schema_change" , "(x int) WITH (checkpoint_interval = 2)" )) {
443+ Path tableLocation = Path .of (getTableLocation (table .getName ()).replace ("file://" , "" ));
444+
445+ assertUpdate ("INSERT INTO " + table .getName () + " VALUES 1" , 1 );
446+ // generate checkpoint at version 2
447+ assertUpdate ("INSERT INTO " + table .getName () + " VALUES 2" , 1 );
448+ assertThat (query ("TABLE " + table .getName ()))
449+ .matches ("VALUES 1, 2" );
450+ assertThat (Files .exists (tableLocation .resolve ("_delta_log/00000000000000000002.checkpoint.parquet" ))).isTrue ();
451+
452+ // alter table to add a new column at version 3
453+ assertUpdate ("CREATE OR REPLACE TABLE " + table .getName () + " WITH (checkpoint_interval = 20) AS SELECT 3 x, CAST('version3' AS varchar) y" , 1 );
454+ // alter table to add a new column not treated as schema change, thus no new checkpoint should be created
455+ assertThat (Files .exists (tableLocation .resolve ("_delta_log/00000000000000000003.checkpoint.parquet" ))).isFalse ();
456+ assertThat (query ("TABLE " + table .getName ()))
457+ .matches ("VALUES (3, varchar 'version3')" );
458+
459+ // alter table to drop a column at version 4
460+ assertUpdate ("CREATE OR REPLACE TABLE " + table .getName () + " WITH (checkpoint_interval = 20) AS SELECT CAST('version4' AS varchar) y" , 1 );
461+ // alter table to drop a column treated as schema change, thus new checkpoint should be created
462+ assertThat (Files .exists (tableLocation .resolve ("_delta_log/00000000000000000004.checkpoint.parquet" ))).isTrue ();
463+ assertThat (query ("TABLE " + table .getName ()))
464+ .matches ("VALUES varchar 'version4'" );
465+
466+ // alter table to rename a column at version 5
467+ assertUpdate ("CREATE OR REPLACE TABLE " + table .getName () + " WITH (checkpoint_interval = 20) AS SELECT CAST('version5' AS varchar) z" , 1 );
468+ // alter table to rename a column treated as schema change, thus new checkpoint should be created
469+ assertThat (Files .exists (tableLocation .resolve ("_delta_log/00000000000000000005.checkpoint.parquet" ))).isTrue ();
470+ assertThat (query ("TABLE " + table .getName ()))
471+ .matches ("VALUES varchar 'version5'" );
472+
473+ // alter table to change column type at version 6
474+ assertUpdate ("CREATE OR REPLACE TABLE " + table .getName () + " WITH (checkpoint_interval = 20) AS SELECT 6 z" , 1 );
475+ // alter table to change column type treated as schema change, thus new checkpoint should be created
476+ assertThat (Files .exists (tableLocation .resolve ("_delta_log/00000000000000000006.checkpoint.parquet" ))).isTrue ();
477+ assertThat (query ("TABLE " + table .getName ()))
478+ .matches ("VALUES 6" );
479+
480+ // alter table to change column type back at version 7, test change type again
481+ assertUpdate ("CREATE OR REPLACE TABLE " + table .getName () + " WITH (checkpoint_interval = 20) AS SELECT CAST('version7' AS varchar) z" , 1 );
482+ assertThat (Files .exists (tableLocation .resolve ("_delta_log/00000000000000000007.checkpoint.parquet" ))).isTrue ();
483+ assertThat (query ("TABLE " + table .getName ()))
484+ .matches ("VALUES varchar 'version7'" );
485+ }
486+ }
487+
340488 @ Test // regression test for https://github.com/trinodb/trino/issues/24121
341489 void testPartitionValuesParsedCheckpoint ()
342490 throws Exception
0 commit comments