Remove support create table with location

docs/src/main/sphinx/connector/delta-lake.md

@@ -519,27 +519,16 @@ When Delta Lake tables exist in storage but not in the metastore, Trino can be
 used to register the tables:
 
 ```
-CREATE TABLE example.default.example_table (
-  dummy BIGINT
-)
-WITH (
-  location = '...'
-)
+CALL example.system.register_table(schema_name => 'testdb', table_name => 'example_table', table_location => 's3://my-bucket/a/path')
 ```
 
-Columns listed in the DDL, such as `dummy` in the preceding example, are
-ignored. The table schema is read from the transaction log instead. If the
+The table schema is read from the transaction log instead. If the
 schema is changed by an external system, Trino automatically uses the new
 schema.
 
 :::{warning}
-Using `CREATE TABLE` with an existing table content is deprecated, instead
-use the `system.register_table` procedure. The `CREATE TABLE ... WITH
-(location=...)` syntax can be temporarily re-enabled using the
-`delta.legacy-create-table-with-existing-location.enabled` catalog
-configuration property or
-`legacy_create_table_with_existing_location_enabled` catalog session
-property.
+Using ``CREATE TABLE`` with an existing table content is disallowed,
+use the ``system.register_table`` procedure instead.
 :::
 
 If the specified location does not already contain a Delta table, the connector

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

@@ -36,7 +36,9 @@
 import static java.util.concurrent.TimeUnit.DAYS;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
-@DefunctConfig("delta.experimental.ignore-checkpoint-write-failures")
+@DefunctConfig({
+        "delta.experimental.ignore-checkpoint-write-failures",
+        "delta.legacy-create-table-with-existing-location.enabled"})
 public class DeltaLakeConfig
 {
     public static final String EXTENDED_STATISTICS_ENABLED = "delta.extended-statistics.enabled";
@@ -74,7 +76,6 @@ public class DeltaLakeConfig
     private String parquetTimeZone = TimeZone.getDefault().getID();
     private DataSize targetMaxFileSize = DataSize.of(1, GIGABYTE);
     private boolean uniqueTableLocation = true;
-    private boolean legacyCreateTableWithExistingLocationEnabled;
     private boolean registerTableProcedureEnabled;
     private boolean projectionPushdownEnabled = true;
     private boolean queryPartitionFilterRequired;
@@ -449,21 +450,6 @@ public DeltaLakeConfig setUniqueTableLocation(boolean uniqueTableLocation)
         return this;
     }
 
-    @Deprecated
-    public boolean isLegacyCreateTableWithExistingLocationEnabled()
-    {
-        return legacyCreateTableWithExistingLocationEnabled;
-    }
-
-    @Deprecated
-    @Config("delta.legacy-create-table-with-existing-location.enabled")
-    @ConfigDescription("Enable using the CREATE TABLE statement to register an existing table")
-    public DeltaLakeConfig setLegacyCreateTableWithExistingLocationEnabled(boolean legacyCreateTableWithExistingLocationEnabled)
-    {
-        this.legacyCreateTableWithExistingLocationEnabled = legacyCreateTableWithExistingLocationEnabled;
-        return this;
-    }
-
     public boolean isRegisterTableProcedureEnabled()
     {
         return registerTableProcedureEnabled;

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

@@ -199,7 +199,6 @@
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.getHiveCatalogName;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isCollectExtendedStatisticsColumnStatisticsOnWrite;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isExtendedStatisticsEnabled;
-import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isLegacyCreateTableWithExistingLocationEnabled;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isProjectionPushdownEnabled;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isQueryPartitionFilterRequired;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isTableStatisticsEnabled;
@@ -966,13 +965,9 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
                 transactionLogWriter.flush();
             }
             else {
-                if (!isLegacyCreateTableWithExistingLocationEnabled(session)) {
-                    throw new TrinoException(
-                            NOT_SUPPORTED,
-                            "Using CREATE TABLE with an existing table content is deprecated, instead use the system.register_table() procedure." +
-                                    " The CREATE TABLE syntax can be temporarily re-enabled using the 'delta.legacy-create-table-with-existing-location.enabled' config property" +
-                                    " or 'legacy_create_table_with_existing_location_enabled' session property.");
-                }
+                throw new TrinoException(
+                        NOT_SUPPORTED,
+                        "Using CREATE TABLE with an existing table content is disallowed, instead use the system.register_table() procedure.");
             }
         }
         catch (IOException e) {

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

@@ -67,7 +67,6 @@ public final class DeltaLakeSessionProperties
     private static final String TABLE_STATISTICS_ENABLED = "statistics_enabled";
     public static final String EXTENDED_STATISTICS_ENABLED = "extended_statistics_enabled";
     public static final String EXTENDED_STATISTICS_COLLECT_ON_WRITE = "extended_statistics_collect_on_write";
-    public static final String LEGACY_CREATE_TABLE_WITH_EXISTING_LOCATION_ENABLED = "legacy_create_table_with_existing_location_enabled";
     private static final String PROJECTION_PUSHDOWN_ENABLED = "projection_pushdown_enabled";
     private static final String QUERY_PARTITION_FILTER_REQUIRED = "query_partition_filter_required";
 
@@ -172,11 +171,6 @@ public DeltaLakeSessionProperties(
                         "Enable collection (ANALYZE) and use of extended statistics.",
                         deltaLakeConfig.isExtendedStatisticsEnabled(),
                         false),
-                booleanProperty(
-                        LEGACY_CREATE_TABLE_WITH_EXISTING_LOCATION_ENABLED,
-                        "Enable using the CREATE TABLE statement to register an existing table",
-                        deltaLakeConfig.isLegacyCreateTableWithExistingLocationEnabled(),
-                        false),
                 booleanProperty(
                         EXTENDED_STATISTICS_COLLECT_ON_WRITE,
                         "Enables automatic column level extended statistics collection on write",
@@ -281,12 +275,6 @@ public static boolean isExtendedStatisticsEnabled(ConnectorSession session)
         return session.getProperty(EXTENDED_STATISTICS_ENABLED, Boolean.class);
     }
 
-    @Deprecated
-    public static boolean isLegacyCreateTableWithExistingLocationEnabled(ConnectorSession session)
-    {
-        return session.getProperty(LEGACY_CREATE_TABLE_WITH_EXISTING_LOCATION_ENABLED, Boolean.class);
-    }
-
     public static boolean isCollectExtendedStatisticsColumnStatisticsOnWrite(ConnectorSession session)
     {
         return session.getProperty(EXTENDED_STATISTICS_COLLECT_ON_WRITE, Boolean.class);

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

@@ -66,7 +66,6 @@ public void testDefaults()
                 .setPerTransactionMetastoreCacheMaximumSize(1000)
                 .setTargetMaxFileSize(DataSize.of(1, GIGABYTE))
                 .setUniqueTableLocation(true)
-                .setLegacyCreateTableWithExistingLocationEnabled(false)
                 .setRegisterTableProcedureEnabled(false)
                 .setProjectionPushdownEnabled(true)
                 .setQueryPartitionFilterRequired(false));
@@ -103,7 +102,6 @@ public void testExplicitPropertyMappings()
                 .put("delta.parquet.time-zone", nonDefaultTimeZone().getID())
                 .put("delta.target-max-file-size", "2 GB")
                 .put("delta.unique-table-location", "false")
-                .put("delta.legacy-create-table-with-existing-location.enabled", "true")
                 .put("delta.register-table-procedure.enabled", "true")
                 .put("delta.projection-pushdown-enabled", "false")
                 .put("delta.query-partition-filter-required", "true")
@@ -137,7 +135,6 @@ public void testExplicitPropertyMappings()
                 .setPerTransactionMetastoreCacheMaximumSize(500)
                 .setTargetMaxFileSize(DataSize.of(2, GIGABYTE))
                 .setUniqueTableLocation(false)
-                .setLegacyCreateTableWithExistingLocationEnabled(true)
                 .setRegisterTableProcedureEnabled(true)
                 .setProjectionPushdownEnabled(false)
                 .setQueryPartitionFilterRequired(true);

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

@@ -1795,6 +1795,21 @@ public void testSettingChangeDataFeedEnabledProperty()
                 .contains("change_data_feed_enabled = true");
     }
 
+    @Test
+    public void testCreateTableWithExistingLocation()
+    {
+        String tableName = "test_legacy_create_table_" + randomNameSuffix();
+
+        assertQuerySucceeds("CREATE TABLE " + tableName + " AS SELECT 1 as a, 'INDIA' as b, true as c");
+        assertQuery("SELECT * FROM " + tableName, "VALUES (1, 'INDIA', true)");
+
+        String tableLocation = (String) computeScalar("SELECT DISTINCT regexp_replace(\"$path\", '/[^/]*$', '') FROM " + tableName);
+        assertUpdate("CALL system.unregister_table(CURRENT_SCHEMA, '" + tableName + "')");
+
+        assertQueryFails(format("CREATE TABLE %s (dummy int) with (location = '%s')", tableName, tableLocation),
+                ".*Using CREATE TABLE with an existing table content is disallowed.*");
+    }
+
     @Test
     public void testProjectionPushdownOnPartitionedTables()
     {