Add register_table procedure support for delta table

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

@@ -155,6 +155,9 @@ values. Typical usage does not require you to configure them.
     * - ``delta.unique-table-location``
       - Use randomized, unique table locations.
       - ``true``
+    * - ``delta.register-table-procedure.enabled``
+      - Enable to allow users to call the ``register_table`` procedure
+      - ``false``
 
 The following table describes performance tuning catalog properties for the
 connector.
@@ -526,6 +529,13 @@ ignored. 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``
+   config property or ``legacy_create_table_with_existing_location_enabled`` session property.
+
 If the specified location does not already contain a Delta table, the connector
 automatically writes the initial transaction log entries and registers the table
 in the metastore. As a result, any Databricks engine can write to the table::
@@ -560,6 +570,21 @@ The following example uses all three table properties::
   )
   AS SELECT name, comment, regionkey FROM tpch.tiny.nation;
 
+.. _delta-lake-register-table:
+
+Register table
+^^^^^^^^^^^^^^
+
+The connector can register table into the metastore with existing transaction logs and data files.
+
+The ``system.register_table`` procedure allows the caller to register an existing delta lake
+table in the metastore, using its existing transaction logs and data files::
+
+    CALL delta.system.register_table(schema_name => 'testdb', table_name => 'customer_orders', table_location => 's3://my-bucket/a/path')
+
+To prevent unauthorized users from accessing data, this procedure is disabled by default.
+The procedure is enabled only when ``delta.register-table-procedure.enabled`` is set to ``true``.
+
 .. _delta-lake-write-support:
 
 Updating data

plugin/trino-delta-lake/pom.xml

@@ -421,6 +421,7 @@
                                 <exclude>**/TestDeltaLakeSharedGlueMetastoreWithTableRedirections.java</exclude>
                                 <exclude>**/TestDeltaLakeTableWithCustomLocationUsingGlueMetastore.java</exclude>
                                 <exclude>**/TestDeltaLakeRenameToWithGlueMetastore.java</exclude>
+                                <exclude>**/TestDeltaLakeRegisterTableProcedureWithGlue.java</exclude>
                                 <exclude>**/TestDeltaLakeGcsConnectorSmokeTest.java</exclude>
                             </excludes>
                         </configuration>
@@ -471,6 +472,7 @@
                                 <include>**/TestDeltaLakeSharedGlueMetastoreWithTableRedirections.java</include>
                                 <include>**/TestDeltaLakeTableWithCustomLocationUsingGlueMetastore.java</include>
                                 <include>**/TestDeltaLakeRenameToWithGlueMetastore.java</include>
+                                <include>**/TestDeltaLakeRegisterTableProcedureWithGlue.java</include>
                             </includes>
                         </configuration>
                     </plugin>

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

@@ -74,6 +74,8 @@ 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;
 
     public Duration getMetadataCacheTtl()
     {
@@ -431,4 +433,32 @@ public DeltaLakeConfig setUniqueTableLocation(boolean uniqueTableLocation)
         this.uniqueTableLocation = 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;
+    }
+
+    @Config("delta.register-table-procedure.enabled")
+    @ConfigDescription("Allow users to call the register_table procedure")
+    public DeltaLakeConfig setRegisterTableProcedureEnabled(boolean registerTableProcedureEnabled)
+    {
+        this.registerTableProcedureEnabled = registerTableProcedureEnabled;
+        return this;
+    }
 }

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

@@ -163,6 +163,7 @@
 import static io.trino.plugin.deltalake.DeltaLakeErrorCode.DELTA_LAKE_BAD_WRITE;
 import static io.trino.plugin.deltalake.DeltaLakeErrorCode.DELTA_LAKE_INVALID_SCHEMA;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isExtendedStatisticsEnabled;
+import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isLegacyCreateTableWithExistingLocationEnabled;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isTableStatisticsEnabled;
 import static io.trino.plugin.deltalake.DeltaLakeTableProperties.CHECKPOINT_INTERVAL_PROPERTY;
 import static io.trino.plugin.deltalake.DeltaLakeTableProperties.LOCATION_PROPERTY;
@@ -727,21 +728,21 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
                 setRollback(() -> deleteRecursivelyIfExists(new HdfsContext(session), hdfsEnvironment, deltaLogDirectory));
                 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.");
+                }
+            }
         }
         catch (IOException e) {
             throw new TrinoException(DELTA_LAKE_BAD_WRITE, "Unable to access file system for: " + location, e);
         }
 
-        Table.Builder tableBuilder = Table.builder()
-                .setDatabaseName(schemaName)
-                .setTableName(tableName)
-                .setOwner(Optional.of(session.getUser()))
-                .setTableType(external ? EXTERNAL_TABLE.name() : MANAGED_TABLE.name())
-                .setDataColumns(DUMMY_DATA_COLUMNS)
-                .setParameters(deltaTableProperties(session, location, external));
-
-        setDeltaStorageFormat(tableBuilder, location, targetPath);
-        Table table = tableBuilder.build();
+        Table table = buildTable(session, schemaTableName, location, targetPath, external);
 
         PrincipalPrivileges principalPrivileges = buildInitialPrivilegeSet(table.getOwner().orElseThrow());
         metastore.createTable(
@@ -750,6 +751,20 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
                 principalPrivileges);
     }
 
+    public static Table buildTable(ConnectorSession session, SchemaTableName schemaTableName, String location, Path targetPath, boolean isExternal)
+    {
+        Table.Builder tableBuilder = Table.builder()
+                .setDatabaseName(schemaTableName.getSchemaName())
+                .setTableName(schemaTableName.getTableName())
+                .setOwner(Optional.of(session.getUser()))
+                .setTableType(isExternal ? EXTERNAL_TABLE.name() : MANAGED_TABLE.name())
+                .setDataColumns(DUMMY_DATA_COLUMNS)
+                .setParameters(deltaTableProperties(session, location, isExternal));
+
+        setDeltaStorageFormat(tableBuilder, location, targetPath);
+        return tableBuilder.build();
+    }
+
     private static Map<String, String> deltaTableProperties(ConnectorSession session, String location, boolean external)
     {
         ImmutableMap.Builder<String, String> properties = ImmutableMap.<String, String>builder()
@@ -950,15 +965,7 @@ public Optional<ConnectorOutputMetadata> finishCreateTable(
                 .map(dataFileInfoCodec::fromJson)
                 .collect(toImmutableList());
 
-        Table.Builder tableBuilder = Table.builder()
-                .setDatabaseName(schemaName)
-                .setTableName(tableName)
-                .setOwner(Optional.of(session.getUser()))
-                .setTableType(handle.isExternal() ? EXTERNAL_TABLE.name() : MANAGED_TABLE.name())
-                .setDataColumns(DUMMY_DATA_COLUMNS)
-                .setParameters(deltaTableProperties(session, location, handle.isExternal()));
-        setDeltaStorageFormat(tableBuilder, location, getExternalPath(new HdfsContext(session), location));
-        Table table = tableBuilder.build();
+        Table table = buildTable(session, schemaTableName(schemaName, tableName), location, getExternalPath(new HdfsContext(session), location), handle.isExternal());
         // Ensure the table has queryId set. This is relied on for exception handling
         String queryId = session.getQueryId();
         verify(

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

@@ -26,6 +26,7 @@
 import io.trino.plugin.deltalake.metastore.DeltaLakeMetastore;
 import io.trino.plugin.deltalake.procedure.DropExtendedStatsProcedure;
 import io.trino.plugin.deltalake.procedure.OptimizeTableProcedure;
+import io.trino.plugin.deltalake.procedure.RegisterTableProcedure;
 import io.trino.plugin.deltalake.procedure.VacuumProcedure;
 import io.trino.plugin.deltalake.statistics.CachingExtendedStatisticsAccess;
 import io.trino.plugin.deltalake.statistics.CachingExtendedStatisticsAccess.ForCachingExtendedStatisticsAccess;
@@ -144,6 +145,7 @@ public void setup(Binder binder)
         Multibinder<Procedure> procedures = newSetBinder(binder, Procedure.class);
         procedures.addBinding().toProvider(DropExtendedStatsProcedure.class).in(Scopes.SINGLETON);
         procedures.addBinding().toProvider(VacuumProcedure.class).in(Scopes.SINGLETON);
+        procedures.addBinding().toProvider(RegisterTableProcedure.class).in(Scopes.SINGLETON);
 
         Multibinder<TableProcedureMetadata> tableProcedures = newSetBinder(binder, TableProcedureMetadata.class);
         tableProcedures.addBinding().toProvider(OptimizeTableProcedure.class).in(Scopes.SINGLETON);

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

@@ -57,6 +57,7 @@ public final class DeltaLakeSessionProperties
     private static final String DYNAMIC_FILTERING_WAIT_TIMEOUT = "dynamic_filtering_wait_timeout";
     private static final String TABLE_STATISTICS_ENABLED = "statistics_enabled";
     public static final String EXTENDED_STATISTICS_ENABLED = "extended_statistics_enabled";
+    public static final String LEGACY_CREATE_TABLE_WITH_EXISTING_LOCATION_ENABLED = "legacy_create_table_with_existing_location_enabled";
 
     private final List<PropertyMetadata<?>> sessionProperties;
 
@@ -146,6 +147,11 @@ public DeltaLakeSessionProperties(
                         "Use extended statistics collected by ANALYZE",
                         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),
                 enumProperty(
                         COMPRESSION_CODEC,
                         "Compression codec to use when writing new data files",
@@ -230,6 +236,12 @@ 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 HiveCompressionCodec getCompressionCodec(ConnectorSession session)
     {
         return session.getProperty(COMPRESSION_CODEC, HiveCompressionCodec.class);

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

@@ -0,0 +1,151 @@
+/*
+ * Licensed 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 io.trino.plugin.deltalake.procedure;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.filesystem.TrinoFileSystem;
+import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.plugin.deltalake.DeltaLakeConfig;
+import io.trino.plugin.deltalake.DeltaLakeMetadataFactory;
+import io.trino.plugin.deltalake.metastore.DeltaLakeMetastore;
+import io.trino.plugin.hive.metastore.PrincipalPrivileges;
+import io.trino.plugin.hive.metastore.Table;
+import io.trino.spi.TrinoException;
+import io.trino.spi.classloader.ThreadContextClassLoader;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.SchemaNotFoundException;
+import io.trino.spi.connector.SchemaTableName;
+import io.trino.spi.procedure.Procedure;
+import org.apache.hadoop.fs.Path;
+
+import javax.inject.Inject;
+import javax.inject.Provider;
+
+import java.io.IOException;
+import java.lang.invoke.MethodHandle;
+
+import static com.google.common.base.Strings.isNullOrEmpty;
+import static io.trino.plugin.base.util.Procedures.checkProcedureArgument;
+import static io.trino.plugin.deltalake.DeltaLakeErrorCode.DELTA_LAKE_FILESYSTEM_ERROR;
+import static io.trino.plugin.deltalake.DeltaLakeMetadata.buildTable;
+import static io.trino.plugin.deltalake.transactionlog.TransactionLogUtil.getTransactionLogDir;
+import static io.trino.plugin.hive.metastore.MetastoreUtil.buildInitialPrivilegeSet;
+import static io.trino.spi.StandardErrorCode.GENERIC_USER_ERROR;
+import static io.trino.spi.StandardErrorCode.PERMISSION_DENIED;
+import static io.trino.spi.type.VarcharType.VARCHAR;
+import static java.lang.String.format;
+import static java.lang.invoke.MethodHandles.lookup;
+import static java.util.Objects.requireNonNull;
+
+public class RegisterTableProcedure
+        implements Provider<Procedure>
+{
+    private static final MethodHandle REGISTER_TABLE;
+
+    private static final String PROCEDURE_NAME = "register_table";
+    private static final String SYSTEM_SCHEMA = "system";
+
+    private static final String SCHEMA_NAME = "SCHEMA_NAME";
+    private static final String TABLE_NAME = "TABLE_NAME";
+    private static final String TABLE_LOCATION = "TABLE_LOCATION";
+
+    static {
+        try {
+            REGISTER_TABLE = lookup().unreflect(RegisterTableProcedure.class.getMethod("registerTable", ConnectorSession.class, String.class, String.class, String.class));
+        }
+        catch (ReflectiveOperationException e) {
+            throw new AssertionError(e);
+        }
+    }
+
+    private final DeltaLakeMetadataFactory metadataFactory;
+    private final TrinoFileSystemFactory fileSystemFactory;
+    private final boolean registerTableProcedureEnabled;
+
+    @Inject
+    public RegisterTableProcedure(DeltaLakeMetadataFactory metadataFactory, TrinoFileSystemFactory fileSystemFactory, DeltaLakeConfig deltaLakeConfig)
+    {
+        this.metadataFactory = requireNonNull(metadataFactory, "metadataFactory is null");
+        this.fileSystemFactory = requireNonNull(fileSystemFactory, "fileSystemFactory is null");
+        this.registerTableProcedureEnabled = deltaLakeConfig.isRegisterTableProcedureEnabled();
+    }
+
+    @Override
+    public Procedure get()
+    {
+        return new Procedure(
+                SYSTEM_SCHEMA,
+                PROCEDURE_NAME,
+                ImmutableList.of(
+                        new Procedure.Argument(SCHEMA_NAME, VARCHAR),
+                        new Procedure.Argument(TABLE_NAME, VARCHAR),
+                        new Procedure.Argument(TABLE_LOCATION, VARCHAR)),
+                REGISTER_TABLE.bindTo(this));
+    }
+
+    public void registerTable(
+            ConnectorSession clientSession,
+            String schemaName,
+            String tableName,
+            String tableLocation)
+    {
+        try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(getClass().getClassLoader())) {
+            doRegisterTable(
+                    clientSession,
+                    schemaName,
+                    tableName,
+                    tableLocation);
+        }
+    }
+
+    private void doRegisterTable(
+            ConnectorSession session,
+            String schemaName,
+            String tableName,
+            String tableLocation)
+    {
+        if (!registerTableProcedureEnabled) {
+            throw new TrinoException(PERMISSION_DENIED, "register_table procedure is disabled");
+        }
+        checkProcedureArgument(!isNullOrEmpty(schemaName), "schema_name cannot be null or empty");
+        checkProcedureArgument(!isNullOrEmpty(tableName), "table_name cannot be null or empty");
+        checkProcedureArgument(!isNullOrEmpty(tableLocation), "table_location cannot be null or empty");
+
+        SchemaTableName schemaTableName = new SchemaTableName(schemaName, tableName);
+        DeltaLakeMetastore metastore = metadataFactory.create(session.getIdentity()).getMetastore();
+
+        if (metastore.getDatabase(schemaName).isEmpty()) {
+            throw new SchemaNotFoundException(schemaTableName.getSchemaName());
+        }
+
+        TrinoFileSystem fileSystem = fileSystemFactory.create(session);
+        try {
+            Path transactionLogDir = getTransactionLogDir(new Path(tableLocation));
+            if (!fileSystem.listFiles(transactionLogDir.toString()).hasNext()) {
+                throw new TrinoException(GENERIC_USER_ERROR, format("No transaction log found in location %s", transactionLogDir));
+            }
+        }
+        catch (IOException e) {
+            throw new TrinoException(DELTA_LAKE_FILESYSTEM_ERROR, format("Failed checking table location %s", tableLocation), e);
+        }
+
+        Table table = buildTable(session, schemaTableName, tableLocation, new Path(tableLocation), true);
+
+        PrincipalPrivileges principalPrivileges = buildInitialPrivilegeSet(table.getOwner().orElseThrow());
+        metastore.createTable(
+                session,
+                table,
+                principalPrivileges);
+    }
+}

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

@@ -33,10 +33,12 @@ protected HiveMinioDataLake createHiveMinioDataLake()
     }
 
     @Override
-    protected void createTableFromResources(String table, String resourcePath, QueryRunner queryRunner)
+    protected void registerTableFromResources(String table, String resourcePath, QueryRunner queryRunner)
     {
         hiveMinioDataLake.copyResources(resourcePath, table);
-        queryRunner.execute(format("CREATE TABLE %s (dummy int) WITH (location = '%s')",
+        queryRunner.execute(format(
+                "CALL system.register_table('%s', '%s', '%s')",
+                SCHEMA,
                 table,
                 getLocationForTable(bucketName, table)));
     }