Add table_changes(tableName, startVersion, endVersion) function for delta lake

table_changes allows reading cdf entries stream.
startVersion and endVersion are optional.
If startVersion is not provided, function starts reading entries from the first one.
startVersion is exclusive, endVersion is inclusive.

Author: homar

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

@@ -724,6 +724,25 @@ The output of the query has the following columns:
     - ``boolean``
     - Whether or not the operation appended data
 
+Functions
+---------------
+
+The connector exposes following functions:
+
+table_changes
+^^^^^^^^^^^^^
+
+Allows reading CDF entries:
+
+.. code-block:: sql
+
+    SELECT * FROM TABLE(system.table_changes('test_schema.test_table', startVersion, endVersion)
+
+``startVersion`` - type ``BIGINT``, exclusive, changes will be read from ``startVersion + 1``
+
+``endVersion`` - type ``BIGINT``, inclusive, changes will be read up to endVersion
+
+If ``startVersion`` is not provided  entire history will be read.
 
 Performance
 -----------

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

@@ -32,7 +32,9 @@
 import io.trino.spi.connector.SystemTable;
 import io.trino.spi.connector.TableProcedureMetadata;
 import io.trino.spi.eventlistener.EventListener;
+import io.trino.spi.function.FunctionProvider;
 import io.trino.spi.procedure.Procedure;
+import io.trino.spi.ptf.ConnectorTableFunction;
 import io.trino.spi.session.PropertyMetadata;
 import io.trino.spi.transaction.IsolationLevel;
 
@@ -68,6 +70,8 @@ public class DeltaLakeConnector
     // Delta lake is not transactional but we use Trino transaction boundaries to create a per-query
     // caching Hive metastore clients. DeltaLakeTransactionManager is used to store those.
     private final DeltaLakeTransactionManager transactionManager;
+    private final Set<ConnectorTableFunction> tableFunctions;
+    private final FunctionProvider functionProvider;
 
     public DeltaLakeConnector(
             LifeCycleManager lifeCycleManager,
@@ -84,7 +88,9 @@ public DeltaLakeConnector(
             List<PropertyMetadata<?>> analyzeProperties,
             Optional<ConnectorAccessControl> accessControl,
             Set<EventListener> eventListeners,
-            DeltaLakeTransactionManager transactionManager)
+            DeltaLakeTransactionManager transactionManager,
+            Set<ConnectorTableFunction> tableFunctions,
+            FunctionProvider functionProvider)
     {
         this.lifeCycleManager = requireNonNull(lifeCycleManager, "lifeCycleManager is null");
         this.splitManager = requireNonNull(splitManager, "splitManager is null");
@@ -104,6 +110,8 @@ public DeltaLakeConnector(
         this.accessControl = requireNonNull(accessControl, "accessControl is null");
         this.eventListeners = ImmutableSet.copyOf(requireNonNull(eventListeners, "eventListeners is null"));
         this.transactionManager = requireNonNull(transactionManager, "transactionManager is null");
+        this.tableFunctions = requireNonNull(tableFunctions, "tableFunctions is null");
+        this.functionProvider = requireNonNull(functionProvider, "functionProvider is null");
     }
 
     @Override
@@ -223,4 +231,16 @@ public Set<ConnectorCapabilities> getCapabilities()
     {
         return immutableEnumSet(NOT_NULL_COLUMN_CONSTRAINT);
     }
+
+    @Override
+    public Set<ConnectorTableFunction> getTableFunctions()
+    {
+        return tableFunctions;
+    }
+
+    @Override
+    public Optional<FunctionProvider> getFunctionProvider()
+    {
+        return Optional.of(functionProvider);
+    }
 }

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

@@ -0,0 +1,71 @@
+/*
+ * 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;
+
+import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.plugin.deltalake.functions.tablechanges.DeltaLakeTableChangesFunction;
+import io.trino.plugin.deltalake.functions.tablechanges.TableChangesProcessorProvider;
+import io.trino.spi.function.AggregationImplementation;
+import io.trino.spi.function.BoundSignature;
+import io.trino.spi.function.FunctionDependencies;
+import io.trino.spi.function.FunctionId;
+import io.trino.spi.function.FunctionProvider;
+import io.trino.spi.function.InvocationConvention;
+import io.trino.spi.function.ScalarFunctionImplementation;
+import io.trino.spi.function.SchemaFunctionName;
+import io.trino.spi.function.WindowFunctionSupplier;
+import io.trino.spi.ptf.TableFunctionProcessorProvider;
+
+import javax.inject.Inject;
+
+public class DeltaLakeFunctionProvider
+        implements FunctionProvider
+{
+    private final TrinoFileSystemFactory fileSystemFactory;
+    private final DeltaLakeConfig deltaLakeConfig;
+
+    @Inject
+    public DeltaLakeFunctionProvider(TrinoFileSystemFactory fileSystemFactory, DeltaLakeConfig deltaLakeConfig)
+    {
+        this.fileSystemFactory = fileSystemFactory;
+        this.deltaLakeConfig = deltaLakeConfig;
+    }
+
+    @Override
+    public ScalarFunctionImplementation getScalarFunctionImplementation(FunctionId functionId, BoundSignature boundSignature, FunctionDependencies functionDependencies, InvocationConvention invocationConvention)
+    {
+        return null;
+    }
+
+    @Override
+    public AggregationImplementation getAggregationImplementation(FunctionId functionId, BoundSignature boundSignature, FunctionDependencies functionDependencies)
+    {
+        return null;
+    }
+
+    @Override
+    public WindowFunctionSupplier getWindowFunctionSupplier(FunctionId functionId, BoundSignature boundSignature, FunctionDependencies functionDependencies)
+    {
+        return null;
+    }
+
+    @Override
+    public TableFunctionProcessorProvider getTableFunctionProcessorProvider(SchemaFunctionName name)
+    {
+        if (name.equals(new SchemaFunctionName(DeltaLakeTableChangesFunction.SCHEMA_NAME, DeltaLakeTableChangesFunction.NAME))) {
+            return new TableChangesProcessorProvider(fileSystemFactory, deltaLakeConfig);
+        }
+        return null;
+    }
+}

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

@@ -2750,7 +2750,7 @@ private static DeltaLakeColumnHandle toColumnHandle(ColumnMetadata column, Strin
         return toColumnHandle(column, OptionalInt.empty(), physicalName, physicalType, partitionColumns);
     }
 
-    private static DeltaLakeColumnHandle toColumnHandle(ColumnMetadata column, OptionalInt fieldId, String physicalName, Type physicalType, Collection<String> partitionColumns)
+    public static DeltaLakeColumnHandle toColumnHandle(ColumnMetadata column, OptionalInt fieldId, String physicalName, Type physicalType, Collection<String> partitionColumns)
     {
         boolean isPartitionKey = partitionColumns.stream().anyMatch(partition -> partition.equalsIgnoreCase(column.getName()));
         return new DeltaLakeColumnHandle(

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

@@ -63,7 +63,9 @@
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.ConnectorSplitManager;
 import io.trino.spi.connector.TableProcedureMetadata;
+import io.trino.spi.function.FunctionProvider;
 import io.trino.spi.procedure.Procedure;
+import io.trino.spi.ptf.ConnectorTableFunction;
 import io.trino.spi.security.ConnectorIdentity;
 
 import javax.inject.Singleton;
@@ -150,6 +152,9 @@ public void setup(Binder binder)
 
         Multibinder<TableProcedureMetadata> tableProcedures = newSetBinder(binder, TableProcedureMetadata.class);
         tableProcedures.addBinding().toProvider(OptimizeTableProcedure.class).in(Scopes.SINGLETON);
+
+        newSetBinder(binder, ConnectorTableFunction.class).addBinding().toProvider(DeltaLakeTableFunctionsProvider.class).in(Scopes.SINGLETON);
+        binder.bind(FunctionProvider.class).to(DeltaLakeFunctionProvider.class).in(Scopes.SINGLETON);
     }
 
     @Singleton

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

@@ -15,7 +15,10 @@
 
 import com.google.common.collect.ImmutableList;
 import io.airlift.units.DataSize;
+import io.trino.filesystem.TrinoFileSystemFactory;
 import io.trino.plugin.base.classloader.ClassLoaderSafeConnectorSplitSource;
+import io.trino.plugin.deltalake.functions.tablechanges.TableChangesFunctionTableHandle;
+import io.trino.plugin.deltalake.functions.tablechanges.TableChangesSplitSource;
 import io.trino.plugin.deltalake.metastore.DeltaLakeMetastore;
 import io.trino.plugin.deltalake.transactionlog.AddFileEntry;
 import io.trino.plugin.hive.HiveTransactionHandle;
@@ -29,9 +32,11 @@
 import io.trino.spi.connector.Constraint;
 import io.trino.spi.connector.DynamicFilter;
 import io.trino.spi.connector.FixedSplitSource;
+import io.trino.spi.function.SchemaFunctionName;
 import io.trino.spi.predicate.Domain;
 import io.trino.spi.predicate.NullableValue;
 import io.trino.spi.predicate.TupleDomain;
+import io.trino.spi.ptf.ConnectorTableFunctionHandle;
 import io.trino.spi.type.TypeManager;
 
 import javax.inject.Inject;
@@ -57,6 +62,7 @@
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.getDynamicFilteringWaitTimeout;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.getMaxInitialSplitSize;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.getMaxSplitSize;
+import static io.trino.plugin.deltalake.functions.tablechanges.DeltaLakeTableChangesFunction.TABLE_CHANGES_NAME;
 import static io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.extractSchema;
 import static io.trino.plugin.deltalake.transactionlog.TransactionLogParser.deserializePartitionValue;
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -74,13 +80,15 @@ public class DeltaLakeSplitManager
     private final int maxSplitsPerSecond;
     private final int maxOutstandingSplits;
     private final double minimumAssignedSplitWeight;
+    private final TrinoFileSystemFactory fileSystemFactory;
 
     @Inject
     public DeltaLakeSplitManager(
             TypeManager typeManager,
             BiFunction<ConnectorSession, HiveTransactionHandle, DeltaLakeMetastore> metastoreProvider,
             ExecutorService executor,
-            DeltaLakeConfig config)
+            DeltaLakeConfig config,
+            TrinoFileSystemFactory fileSystemFactory)
     {
         this.typeManager = requireNonNull(typeManager, "typeManager is null");
         this.metastoreProvider = requireNonNull(metastoreProvider, "metastoreProvider is null");
@@ -89,6 +97,7 @@ public DeltaLakeSplitManager(
         this.maxSplitsPerSecond = config.getMaxSplitsPerSecond();
         this.maxOutstandingSplits = config.getMaxOutstandingSplits();
         this.minimumAssignedSplitWeight = config.getMinimumAssignedSplitWeight();
+        this.fileSystemFactory = requireNonNull(fileSystemFactory, "fileSystem is null");
     }
 
     @Override
@@ -120,6 +129,15 @@ public ConnectorSplitSource getSplits(
         return new ClassLoaderSafeConnectorSplitSource(splitSource, DeltaLakeSplitManager.class.getClassLoader());
     }
 
+    @Override
+    public ConnectorSplitSource getSplits(ConnectorTransactionHandle transaction, ConnectorSession session, SchemaFunctionName name, ConnectorTableFunctionHandle function)
+    {
+        if (name.equals(TABLE_CHANGES_NAME)) {
+            return new TableChangesSplitSource(session, getMetastore(session, transaction), fileSystemFactory, (TableChangesFunctionTableHandle) function);
+        }
+        return ConnectorSplitManager.super.getSplits(transaction, session, name, function);
+    }
+
     private Stream<DeltaLakeSplit> getSplits(
             ConnectorTransactionHandle transaction,
             DeltaLakeTableHandle tableHandle,

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

@@ -0,0 +1,46 @@
+/*
+ * 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;
+
+import io.trino.plugin.base.classloader.ClassLoaderSafeConnectorTableFunction;
+import io.trino.plugin.deltalake.functions.tablechanges.DeltaLakeTableChangesFunction;
+import io.trino.spi.ptf.ConnectorTableFunction;
+import io.trino.spi.type.TypeManager;
+
+import javax.inject.Inject;
+import javax.inject.Provider;
+
+import static java.util.Objects.requireNonNull;
+
+public class DeltaLakeTableFunctionsProvider
+        implements Provider<ConnectorTableFunction>
+{
+    private final DeltaLakeMetadataFactory deltaLakeMetadataFactory;
+    private final TypeManager typeManager;
+
+    @Inject
+    public DeltaLakeTableFunctionsProvider(DeltaLakeMetadataFactory deltaLakeMetadataFactory, TypeManager typeManager)
+    {
+        this.deltaLakeMetadataFactory = requireNonNull(deltaLakeMetadataFactory, "deltaLakeMetadataFactory is null");
+        this.typeManager = requireNonNull(typeManager, "typeManager is null");
+    }
+
+    @Override
+    public ConnectorTableFunction get()
+    {
+        return new ClassLoaderSafeConnectorTableFunction(
+                new DeltaLakeTableChangesFunction(typeManager, deltaLakeMetadataFactory),
+                getClass().getClassLoader());
+    }
+}

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

@@ -53,7 +53,9 @@
 import io.trino.spi.connector.ConnectorSplitManager;
 import io.trino.spi.connector.TableProcedureMetadata;
 import io.trino.spi.eventlistener.EventListener;
+import io.trino.spi.function.FunctionProvider;
 import io.trino.spi.procedure.Procedure;
+import io.trino.spi.ptf.ConnectorTableFunction;
 import io.trino.spi.type.TypeManager;
 import org.weakref.jmx.guice.MBeanModule;
 
@@ -134,6 +136,9 @@ public static Connector createConnector(
             Set<Procedure> procedures = injector.getInstance(Key.get(new TypeLiteral<Set<Procedure>>() {}));
             Set<TableProcedureMetadata> tableProcedures = injector.getInstance(Key.get(new TypeLiteral<Set<TableProcedureMetadata>>() {}));
 
+            Set<ConnectorTableFunction> connectorTableFunctions = injector.getInstance(Key.get(new TypeLiteral<Set<ConnectorTableFunction>>() {}));
+            FunctionProvider functionProvider = injector.getInstance(FunctionProvider.class);
+
             return new DeltaLakeConnector(
                     lifeCycleManager,
                     new ClassLoaderSafeConnectorSplitManager(splitManager, classLoader),
@@ -149,7 +154,9 @@ public static Connector createConnector(
                     deltaLakeAnalyzeProperties.getAnalyzeProperties(),
                     deltaAccessControl,
                     eventListeners,
-                    transactionManager);
+                    transactionManager,
+                    connectorTableFunctions,
+                    functionProvider);
         }
     }
 }