Add table_changes function for delta lake

table_changes allows reading cdf entries stream.
since_version is optional and exclusive.
If since_version is not provided entire history of the table will be read.

Author: homar

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/util/Functions.java

@@ -0,0 +1,33 @@
+/*
+ * 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.base.util;
+
+import com.google.errorprone.annotations.FormatMethod;
+import io.trino.spi.TrinoException;
+
+import static io.trino.spi.StandardErrorCode.INVALID_FUNCTION_ARGUMENT;
+import static java.lang.String.format;
+
+public final class Functions
+{
+    private Functions() {}
+
+    @FormatMethod
+    public static void checkFunctionArgument(boolean condition, String message, Object... args)
+    {
+        if (!condition) {
+            throw new TrinoException(INVALID_FUNCTION_ARGUMENT, format(message, args));
+        }
+    }
+}

plugin/trino-delta-lake/pom.xml

@@ -196,6 +196,12 @@
             <scope>runtime</scope>
         </dependency>
 
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-memory-context</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+
         <dependency>
             <groupId>io.airlift</groupId>
             <artifactId>log-manager</artifactId>

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

@@ -185,6 +185,13 @@ public boolean isBaseColumn()
         return projectionInfo.isEmpty();
     }
 
+    @JsonIgnore
+    public Type getType()
+    {
+        return projectionInfo.map(DeltaLakeColumnProjectionInfo::getType)
+                .orElse(baseType);
+    }
+
     @Override
     public int hashCode()
     {

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 = ImmutableSet.copyOf(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,45 @@
+/*
+ * 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.deltalake.functions.tablechanges.TableChangesProcessorProvider;
+import io.trino.plugin.deltalake.functions.tablechanges.TableChangesTableFunctionHandle;
+import io.trino.spi.function.FunctionProvider;
+import io.trino.spi.ptf.ConnectorTableFunctionHandle;
+import io.trino.spi.ptf.TableFunctionProcessorProvider;
+
+import javax.inject.Inject;
+
+import static java.util.Objects.requireNonNull;
+
+public class DeltaLakeFunctionProvider
+        implements FunctionProvider
+{
+    private final TableChangesProcessorProvider tableChangesProcessorProvider;
+
+    @Inject
+    public DeltaLakeFunctionProvider(TableChangesProcessorProvider tableChangesProcessorProvider)
+    {
+        this.tableChangesProcessorProvider = requireNonNull(tableChangesProcessorProvider, "tableChangesProcessorProvider is null");
+    }
+
+    @Override
+    public TableFunctionProcessorProvider getTableFunctionProcessorProvider(ConnectorTableFunctionHandle functionHandle)
+    {
+        if (functionHandle instanceof TableChangesTableFunctionHandle) {
+            return tableChangesProcessorProvider;
+        }
+        throw new UnsupportedOperationException("Unsupported function: " + functionHandle);
+    }
+}

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

@@ -288,6 +288,8 @@ public class DeltaLakeMetadata
     public static final String ADD_COLUMN_OPERATION = "ADD COLUMNS";
     public static final String INSERT_OPERATION = "WRITE";
     public static final String MERGE_OPERATION = "MERGE";
+    public static final String UPDATE_OPERATION = "UPDATE"; // used by old Trino versions and Spark
+    public static final String DELETE_OPERATION = "DELETE"; // used by old Trino versions and Spark
     public static final String OPTIMIZE_OPERATION = "OPTIMIZE";
     public static final String SET_TBLPROPERTIES_OPERATION = "SET TBLPROPERTIES";
     public static final String CHANGE_COLUMN_OPERATION = "CHANGE COLUMN";

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

@@ -23,6 +23,8 @@
 import io.trino.plugin.base.CatalogName;
 import io.trino.plugin.base.security.ConnectorAccessControlModule;
 import io.trino.plugin.base.session.SessionPropertiesProvider;
+import io.trino.plugin.deltalake.functions.tablechanges.TableChangesFunctionProvider;
+import io.trino.plugin.deltalake.functions.tablechanges.TableChangesProcessorProvider;
 import io.trino.plugin.deltalake.procedure.DropExtendedStatsProcedure;
 import io.trino.plugin.deltalake.procedure.FlushMetadataCacheProcedure;
 import io.trino.plugin.deltalake.procedure.OptimizeTableProcedure;
@@ -57,7 +59,9 @@
 import io.trino.spi.connector.ConnectorPageSourceProvider;
 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 javax.inject.Singleton;
 
@@ -141,6 +145,10 @@ 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(TableChangesFunctionProvider.class).in(Scopes.SINGLETON);
+        binder.bind(FunctionProvider.class).to(DeltaLakeFunctionProvider.class).in(Scopes.SINGLETON);
+        binder.bind(TableChangesProcessorProvider.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.TableChangesSplitSource;
+import io.trino.plugin.deltalake.functions.tablechanges.TableChangesTableFunctionHandle;
 import io.trino.plugin.deltalake.transactionlog.AddFileEntry;
 import io.trino.plugin.deltalake.transactionlog.TableSnapshot;
 import io.trino.plugin.deltalake.transactionlog.TransactionLogAccess;
@@ -33,6 +36,7 @@
 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;
@@ -76,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,
             TransactionLogAccess transactionLogAccess,
             ExecutorService executor,
-            DeltaLakeConfig config)
+            DeltaLakeConfig config,
+            TrinoFileSystemFactory fileSystemFactory)
     {
         this.typeManager = requireNonNull(typeManager, "typeManager is null");
         this.transactionLogAccess = requireNonNull(transactionLogAccess, "transactionLogAccess is null");
@@ -91,6 +97,7 @@ public DeltaLakeSplitManager(
         this.maxSplitsPerSecond = config.getMaxSplitsPerSecond();
         this.maxOutstandingSplits = config.getMaxOutstandingSplits();
         this.minimumAssignedSplitWeight = config.getMinimumAssignedSplitWeight();
+        this.fileSystemFactory = requireNonNull(fileSystemFactory, "fileSystemFactory is null");
     }
 
     @Override
@@ -122,6 +129,15 @@ public ConnectorSplitSource getSplits(
         return new ClassLoaderSafeConnectorSplitSource(splitSource, DeltaLakeSplitManager.class.getClassLoader());
     }
 
+    @Override
+    public ConnectorSplitSource getSplits(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorTableFunctionHandle function)
+    {
+        if (function instanceof TableChangesTableFunctionHandle tableFunctionHandle) {
+            return new TableChangesSplitSource(session, fileSystemFactory, tableFunctionHandle);
+        }
+        throw new UnsupportedOperationException("Unrecognized function: " + function);
+    }
+
     private Stream<DeltaLakeSplit> getSplits(
             DeltaLakeTableHandle tableHandle,
             ConnectorSession session,

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

@@ -56,7 +56,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;
 
@@ -142,6 +144,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),
@@ -157,7 +162,9 @@ public static Connector createConnector(
                     deltaLakeAnalyzeProperties.getAnalyzeProperties(),
                     deltaAccessControl,
                     eventListeners,
-                    transactionManager);
+                    transactionManager,
+                    connectorTableFunctions,
+                    functionProvider);
         }
     }
 }

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/functions/tablechanges/TableChangesFileType.java

@@ -0,0 +1,20 @@
+/*
+ * 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.functions.tablechanges;
+
+public enum TableChangesFileType
+{
+    DATA_FILE,
+    CDF_FILE,
+}