Add support for row-level modification in Cassandra connector

codluca · codluca · commit ce1cf50f1e07 · 2025-09-21T21:00:56.000+03:00
The WHERE clause in an UPDATE statement for Cassandra must specify the full primary key (all columns in the partition key and all columns in the clustering key, if any). The conditions in the WHERE clause must be EQ conditions.
diff --git a/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java b/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java
@@ -53,11 +53,20 @@ public TupleDomain<ColumnHandle> getUnenforcedConstraints()
         return predicates.filter((columnHandle, domain) -> !clusteringPushDownResult.hasBeenFullyPushed(columnHandle));
     }
 
+    public boolean includesAllClusteringColumnsAndHasAllEqPredicates()
+    {
+        return clusteringPushDownResult.includesAllClusteringColumnsAndHasAllEqPredicates;
+    }
+
     private ClusteringPushDownResult getClusteringKeysSet(List<CassandraColumnHandle> clusteringColumns, TupleDomain<ColumnHandle> predicates)
     {
         ImmutableSet.Builder<ColumnHandle> fullyPushedColumnPredicates = ImmutableSet.builder();
         ImmutableList.Builder<String> clusteringColumnSql = ImmutableList.builder();
+        int eqPredicatesCount = 0;
         for (CassandraColumnHandle columnHandle : clusteringColumns) {
+            if (predicates.getDomains().isEmpty()) {
+                break;
+            }
             Domain domain = predicates.getDomains().get().get(columnHandle);
             if (domain == null) {
                 break;
@@ -109,10 +118,13 @@ private ClusteringPushDownResult getClusteringKeysSet(List<CassandraColumnHandle
             if (predicateString.contains(">") || predicateString.contains("<")) {
                 break;
             }
+            else {
+                eqPredicatesCount++;
+            }
         }
         List<String> clusteringColumnPredicates = clusteringColumnSql.build();
 
-        return new ClusteringPushDownResult(fullyPushedColumnPredicates.build(), Joiner.on(" AND ").join(clusteringColumnPredicates));
+        return new ClusteringPushDownResult(fullyPushedColumnPredicates.build(), Joiner.on(" AND ").join(clusteringColumnPredicates), eqPredicatesCount == clusteringColumns.size());
     }
 
     private String toCqlLiteral(CassandraColumnHandle columnHandle, Object value)
@@ -163,7 +175,7 @@ private String translateRangeIntoCql(CassandraColumnHandle columnHandle, Range r
         return upperBoundPredicate;
     }
 
-    private record ClusteringPushDownResult(Set<ColumnHandle> fullyPushedColumnPredicates, String domainQuery)
+    private record ClusteringPushDownResult(Set<ColumnHandle> fullyPushedColumnPredicates, String domainQuery, boolean includesAllClusteringColumnsAndHasAllEqPredicates)
     {
         private ClusteringPushDownResult
         {
diff --git a/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/CassandraMetadata.java b/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/CassandraMetadata.java
@@ -42,11 +42,13 @@
 import io.trino.spi.connector.RelationColumnsMetadata;
 import io.trino.spi.connector.RelationCommentMetadata;
 import io.trino.spi.connector.RetryMode;
+import io.trino.spi.connector.RowChangeParadigm;
 import io.trino.spi.connector.SaveMode;
 import io.trino.spi.connector.SchemaNotFoundException;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.connector.TableFunctionApplicationResult;
 import io.trino.spi.connector.TableNotFoundException;
+import io.trino.spi.expression.Constant;
 import io.trino.spi.function.table.ConnectorTableFunctionHandle;
 import io.trino.spi.predicate.TupleDomain;
 import io.trino.spi.statistics.ComputedStatistics;
@@ -78,6 +80,7 @@
 import static io.trino.spi.connector.RelationColumnsMetadata.forTable;
 import static io.trino.spi.connector.RelationCommentMetadata.forRelation;
 import static io.trino.spi.connector.RetryMode.NO_RETRIES;
+import static io.trino.spi.connector.RowChangeParadigm.CHANGE_ONLY_UPDATED_COLUMNS;
 import static io.trino.spi.connector.SaveMode.REPLACE;
 import static java.lang.String.format;
 import static java.util.Locale.ENGLISH;
@@ -285,6 +288,7 @@ public Optional<ConstraintApplicationResult<ConnectorTableHandle>> applyFilter(C
 
         String clusteringKeyPredicates = "";
         TupleDomain<ColumnHandle> unenforcedConstraint;
+        Boolean includesAllClusteringColumnsAndHasAllEqPredicates = null;
         if (partitionResult.unpartitioned() || partitionResult.indexedColumnPredicatePushdown()) {
             // When the filter is missing at least one of the partition keys or when the table is not partitioned,
             // use the raw unenforced constraint of the partitionResult
@@ -297,6 +301,7 @@ public Optional<ConstraintApplicationResult<ConnectorTableHandle>> applyFilter(C
                     partitionResult.unenforcedConstraint());
             clusteringKeyPredicates = clusteringPredicatesExtractor.getClusteringKeyPredicates();
             unenforcedConstraint = clusteringPredicatesExtractor.getUnenforcedConstraints();
+            includesAllClusteringColumnsAndHasAllEqPredicates = clusteringPredicatesExtractor.includesAllClusteringColumnsAndHasAllEqPredicates();
         }
 
         Optional<List<CassandraPartition>> currentPartitions = handle.getPartitions();
@@ -313,7 +318,8 @@ public Optional<ConstraintApplicationResult<ConnectorTableHandle>> applyFilter(C
                         handle.getTableName(),
                         Optional.of(partitionResult.partitions()),
                         // TODO this should probably be AND-ed with handle.getClusteringKeyPredicates()
-                        clusteringKeyPredicates)),
+                        clusteringKeyPredicates,
+                        includesAllClusteringColumnsAndHasAllEqPredicates)),
                         unenforcedConstraint,
                         constraint.getExpression(),
                         false));
@@ -470,39 +476,103 @@ private static boolean isHiddenIdColumn(CassandraColumnHandle columnHandle)
     }
 
     @Override
-    public ConnectorMergeTableHandle beginMerge(ConnectorSession session, ConnectorTableHandle tableHandle, Map<Integer, Collection<ColumnHandle>> updateCaseColumns, RetryMode retryMode)
+    public RowChangeParadigm getRowChangeParadigm(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
-        throw new TrinoException(NOT_SUPPORTED, "Delete without primary key or partition key is not supported");
+        return CHANGE_ONLY_UPDATED_COLUMNS;
     }
 
+    /**
+     * Attempt to push down an update operation into the connector. If a connector
+     * can execute an update for the table handle on its own, it should return a
+     * table handle, which will be passed back to {@link #executeUpdate} during
+     * query executing to actually execute the update.
+     */
     @Override
-    public ColumnHandle getMergeRowIdColumnHandle(ConnectorSession session, ConnectorTableHandle tableHandle)
+    public Optional<ConnectorTableHandle> applyUpdate(ConnectorSession session, ConnectorTableHandle tableHandle, Map<ColumnHandle, Constant> assignments)
     {
-        return new CassandraColumnHandle("$update_row_id", 0, CassandraTypes.TEXT, false, false, false, true);
+        CassandraNamedRelationHandle table = ((CassandraTableHandle) tableHandle).getRequiredNamedRelation();
+        if (cassandraSession.isMaterializedView(table.getSchemaTableName())) {
+            throw new TrinoException(NOT_SUPPORTED, "Updating materialized views not yet supported");
+        }
+
+        CassandraNamedRelationHandle handle = ((CassandraTableHandle) tableHandle).getRequiredNamedRelation();
+        List<CassandraPartition> partitions = handle.getPartitions()
+                .orElseThrow(() -> new TrinoException(NOT_SUPPORTED, "Updating without partition key is not supported"));
+        if (partitions.isEmpty()) {
+            // there are no records of a given partition key
+            throw new TrinoException(NOT_SUPPORTED, "Updating without partition key is not supported");
+        }
+        if (!handle.getIncludesAllClusteringColumnsAndHasAllEqPredicates()) {
+            throw new TrinoException(NOT_SUPPORTED, "Updating without all clustering keys or with non-eq predicates is not supported");
+        }
+
+        Map<String, String> assignmentsMap = new HashMap<>();
+        for (Map.Entry<ColumnHandle, Constant> entry : assignments.entrySet()) {
+            CassandraColumnHandle column = (CassandraColumnHandle) entry.getKey();
+            if (isHiddenIdColumn(column)) {
+                throw new TrinoException(NOT_SUPPORTED, "Updating the hidden id column is not supported");
+            }
+            Object value = entry.getValue().getValue();
+            if (value == null) {
+                throw new TrinoException(NOT_SUPPORTED, "Updating columns to null is not supported");
+            }
+            String cqlLiteral = cassandraTypeManager.toCqlLiteral(column.cassandraType(), value); // validate that the value can be converted to the cassandra type
+            assignmentsMap.put(column.name(), cqlLiteral);
+        }
+
+        return Optional.of(new CassandraUpdateTableHandle(tableHandle, assignmentsMap));
     }
 
+    /**
+     * Execute the update operation on the handle returned from {@link #applyUpdate}.
+     */
     @Override
-    public Optional<ConnectorTableHandle> applyDelete(ConnectorSession session, ConnectorTableHandle handle)
+    public OptionalLong executeUpdate(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
-        return Optional.of(handle);
+        CassandraUpdateTableHandle updateTableHandle = ((CassandraUpdateTableHandle) tableHandle);
+        CassandraTableHandle cassandraTableHandle = (CassandraTableHandle) updateTableHandle.tableHandle();
+        CassandraNamedRelationHandle namedRelationHandle = cassandraTableHandle.getRequiredNamedRelation();
+        for (String cql : CassandraCqlUtils.getUpdateQueries(namedRelationHandle, updateTableHandle.assignments())) {
+            cassandraSession.execute(cql);
+        }
+        return OptionalLong.empty();
     }
 
     @Override
-    public OptionalLong executeDelete(ConnectorSession session, ConnectorTableHandle deleteHandle)
+    public ColumnHandle getMergeRowIdColumnHandle(ConnectorSession session, ConnectorTableHandle tableHandle)
+    {
+        return new CassandraColumnHandle("$update_row_id", 0, CassandraTypes.TEXT, false, false, false, true);
+    }
+
+    @Override
+    public Optional<ConnectorTableHandle> applyDelete(ConnectorSession session, ConnectorTableHandle deleteHandle)
     {
         CassandraNamedRelationHandle handle = ((CassandraTableHandle) deleteHandle).getRequiredNamedRelation();
         List<CassandraPartition> partitions = handle.getPartitions()
                 .orElseThrow(() -> new TrinoException(NOT_SUPPORTED, "Deleting without partition key is not supported"));
         if (partitions.isEmpty()) {
             // there are no records of a given partition key
-            return OptionalLong.empty();
+            return Optional.empty();
         }
+        return Optional.of(deleteHandle);
+    }
+
+    @Override
+    public OptionalLong executeDelete(ConnectorSession session, ConnectorTableHandle deleteHandle)
+    {
+        CassandraNamedRelationHandle handle = ((CassandraTableHandle) deleteHandle).getRequiredNamedRelation();
         for (String cql : CassandraCqlUtils.getDeleteQueries(handle)) {
             cassandraSession.execute(cql);
         }
         return OptionalLong.empty();
     }
 
+    @Override
+    public ConnectorMergeTableHandle beginMerge(ConnectorSession session, ConnectorTableHandle tableHandle, Map<Integer, Collection<ColumnHandle>> updateCaseColumns, RetryMode retryMode)
+    {
+        throw new TrinoException(NOT_SUPPORTED, "This connector does not support modifying table rows");
+    }
+
     @Override
     public Optional<TableFunctionApplicationResult<ConnectorTableHandle>> applyTableFunction(ConnectorSession session, ConnectorTableFunctionHandle handle)
     {
diff --git a/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/CassandraNamedRelationHandle.java b/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/CassandraNamedRelationHandle.java
@@ -34,23 +34,26 @@ public class CassandraNamedRelationHandle
     private final String tableName;
     private final Optional<List<CassandraPartition>> partitions;
     private final String clusteringKeyPredicates;
+    private final Boolean includesAllClusteringColumnsAndHasAllEqPredicates;
 
     public CassandraNamedRelationHandle(String schemaName, String tableName)
     {
-        this(schemaName, tableName, Optional.empty(), "");
+        this(schemaName, tableName, Optional.empty(), "", null);
     }
 
     @JsonCreator
     public CassandraNamedRelationHandle(
             @JsonProperty("schemaName") String schemaName,
             @JsonProperty("tableName") String tableName,
             @JsonProperty("partitions") Optional<List<CassandraPartition>> partitions,
-            @JsonProperty("clusteringKeyPredicates") String clusteringKeyPredicates)
+            @JsonProperty("clusteringKeyPredicates") String clusteringKeyPredicates,
+            @JsonProperty("includesAllClusteringColumnsAndHasAllEqPredicates") Boolean includesAllClusteringColumnsAndHasAllEqPredicates)
     {
         this.schemaName = requireNonNull(schemaName, "schemaName is null");
         this.tableName = requireNonNull(tableName, "tableName is null");
         this.partitions = partitions.map(ImmutableList::copyOf);
         this.clusteringKeyPredicates = requireNonNull(clusteringKeyPredicates, "clusteringKeyPredicates is null");
+        this.includesAllClusteringColumnsAndHasAllEqPredicates = includesAllClusteringColumnsAndHasAllEqPredicates;
     }
 
     @JsonProperty
@@ -77,6 +80,12 @@ public String getClusteringKeyPredicates()
         return clusteringKeyPredicates;
     }
 
+    @JsonProperty
+    public Boolean getIncludesAllClusteringColumnsAndHasAllEqPredicates()
+    {
+        return includesAllClusteringColumnsAndHasAllEqPredicates;
+    }
+
     public SchemaTableName getSchemaTableName()
     {
         return new SchemaTableName(schemaName, tableName);
@@ -101,7 +110,8 @@ public boolean equals(Object obj)
         return Objects.equals(this.schemaName, other.schemaName) &&
                 Objects.equals(this.tableName, other.tableName) &&
                 Objects.equals(this.partitions, other.partitions) &&
-                Objects.equals(this.clusteringKeyPredicates, other.clusteringKeyPredicates);
+                Objects.equals(this.clusteringKeyPredicates, other.clusteringKeyPredicates) &&
+                Objects.equals(this.includesAllClusteringColumnsAndHasAllEqPredicates, other.includesAllClusteringColumnsAndHasAllEqPredicates);
     }
 
     @Override
@@ -122,6 +132,9 @@ public String toString()
         if (!clusteringKeyPredicates.isEmpty()) {
             result += format(" constraint(%s)", clusteringKeyPredicates);
         }
+        if (includesAllClusteringColumnsAndHasAllEqPredicates != null) {
+            result += format(" includesAllClusteringColumnsAndHasAllEqPredicates(%s)", includesAllClusteringColumnsAndHasAllEqPredicates);
+        }
         return result;
     }
 }
diff --git a/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/CassandraUpdateTableHandle.java b/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/CassandraUpdateTableHandle.java
@@ -0,0 +1,38 @@
+/*
+ * 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.cassandra;
+
+import io.trino.spi.connector.ConnectorTableHandle;
+
+import java.util.Map;
+
+import static java.util.Objects.requireNonNull;
+
+public record CassandraUpdateTableHandle(
+        ConnectorTableHandle tableHandle,
+        Map<String, String> assignments)
+        implements ConnectorTableHandle
+{
+    public CassandraUpdateTableHandle
+    {
+        requireNonNull(tableHandle, "tableHandle is null");
+        requireNonNull(assignments, "assignments is null");
+    }
+
+    @Override
+    public String toString()
+    {
+        return "cassandra:" + tableHandle;
+    }
+}
diff --git a/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/util/CassandraCqlUtils.java b/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/util/CassandraCqlUtils.java
@@ -24,6 +24,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 import static com.datastax.oss.driver.internal.core.util.Strings.doubleQuote;
 import static com.google.common.collect.ImmutableList.toImmutableList;
@@ -113,6 +114,33 @@ private static String getWhereCondition(String partition, String clusteringKeyPr
         return String.join(" AND ", conditions);
     }
 
+    private static String getSetStatements(Map<String, String> assignments)
+    {
+        if (!assignments.isEmpty()) {
+            List<String> setStatements = new ArrayList<>();
+            for (Map.Entry<String, String> entry : assignments.entrySet()) {
+                setStatements.add(format("%s = %s", validColumnName(entry.getKey()), entry.getValue()));
+            }
+            return String.join(" AND ", setStatements);
+        }
+        return "";
+    }
+
+    private static String update(String schemaName, String tableName, CassandraPartition partition, String clusteringKeyPredicates, Map<String, String> assignments)
+    {
+        return format("UPDATE \"%s\".\"%s\" SET %s WHERE %s",
+                schemaName, tableName, getSetStatements(assignments), getWhereCondition(partition.getPartitionId(), clusteringKeyPredicates));
+    }
+
+    public static List<String> getUpdateQueries(CassandraNamedRelationHandle handle, Map<String, String> assignments)
+    {
+        ImmutableList.Builder<String> queries = ImmutableList.builder();
+        for (CassandraPartition partition : handle.getPartitions().orElse(ImmutableList.of())) {
+            queries.add(update(handle.getSchemaName(), handle.getTableName(), partition, handle.getClusteringKeyPredicates(), assignments));
+        }
+        return queries.build();
+    }
+
     private static String deleteFrom(String schemaName, String tableName, CassandraPartition partition, String clusteringKeyPredicates)
     {
         return format("DELETE FROM \"%s\".\"%s\" WHERE %s",
diff --git a/plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestCassandraConnector.java b/plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestCassandraConnector.java
@@ -462,7 +462,7 @@ private static Map<String, Integer> indexColumns(List<ColumnHandle> columnHandle
     private CassandraTableHandle getTableHandle(Optional<List<CassandraPartition>> partitions, String clusteringKeyPredicates)
     {
         CassandraNamedRelationHandle handle = ((CassandraTableHandle) getTableHandle(tableForDelete)).getRequiredNamedRelation();
-        return new CassandraTableHandle(new CassandraNamedRelationHandle(handle.getSchemaName(), handle.getTableName(), partitions, clusteringKeyPredicates));
+        return new CassandraTableHandle(new CassandraNamedRelationHandle(handle.getSchemaName(), handle.getTableName(), partitions, clusteringKeyPredicates, null));
     }
 
     private CassandraPartition createPartition(long value1, long value2)
diff --git a/plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestCassandraConnectorTest.java b/plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestCassandraConnectorTest.java
diff --git a/plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestCassandraSplitManager.java b/plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestCassandraSplitManager.java
diff --git a/plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestJsonCassandraHandles.java b/plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestJsonCassandraHandles.java

Original file line number	Diff line number	Diff line change
`@@ -53,11 +53,20 @@ public TupleDomain<ColumnHandle> getUnenforcedConstraints()`
`53`	`53`	`return predicates.filter((columnHandle, domain) -> !clusteringPushDownResult.hasBeenFullyPushed(columnHandle));`
`54`	`54`	`}`
`55`	`55`
	`56`	`+ public boolean includesAllClusteringColumnsAndHasAllEqPredicates()`
	`57`	`+ {`
	`58`	`+ return clusteringPushDownResult.includesAllClusteringColumnsAndHasAllEqPredicates;`
	`59`	`+ }`
	`60`	`+`
`56`	`61`	`private ClusteringPushDownResult getClusteringKeysSet(List<CassandraColumnHandle> clusteringColumns, TupleDomain<ColumnHandle> predicates)`
`57`	`62`	`{`
`58`	`63`	`ImmutableSet.Builder<ColumnHandle> fullyPushedColumnPredicates = ImmutableSet.builder();`
`59`	`64`	`ImmutableList.Builder<String> clusteringColumnSql = ImmutableList.builder();`
	`65`	`+ int eqPredicatesCount = 0;`
`60`	`66`	`for (CassandraColumnHandle columnHandle : clusteringColumns) {`
	`67`	`+ if (predicates.getDomains().isEmpty()) {`
	`68`	`+ break;`
	`69`	`+ }`
`61`	`70`	`Domain domain = predicates.getDomains().get().get(columnHandle);`
`62`	`71`	`if (domain == null) {`
`63`	`72`	`break;`
`@@ -109,10 +118,13 @@ private ClusteringPushDownResult getClusteringKeysSet(List<CassandraColumnHandle`
`109`	`118`	`if (predicateString.contains(">") \|\| predicateString.contains("<")) {`
`110`	`119`	`break;`
`111`	`120`	`}`
	`121`	`+ else {`
	`122`	`+ eqPredicatesCount++;`
	`123`	`+ }`
`112`	`124`	`}`
`113`	`125`	`List<String> clusteringColumnPredicates = clusteringColumnSql.build();`
`114`	`126`
`115`		`- return new ClusteringPushDownResult(fullyPushedColumnPredicates.build(), Joiner.on(" AND ").join(clusteringColumnPredicates));`
	`127`	`+ return new ClusteringPushDownResult(fullyPushedColumnPredicates.build(), Joiner.on(" AND ").join(clusteringColumnPredicates), eqPredicatesCount == clusteringColumns.size());`
`116`	`128`	`}`
`117`	`129`
`118`	`130`	`private String toCqlLiteral(CassandraColumnHandle columnHandle, Object value)`
`@@ -163,7 +175,7 @@ private String translateRangeIntoCql(CassandraColumnHandle columnHandle, Range r`
`163`	`175`	`return upperBoundPredicate;`
`164`	`176`	`}`
`165`	`177`
`166`		`- private record ClusteringPushDownResult(Set<ColumnHandle> fullyPushedColumnPredicates, String domainQuery)`
	`178`	`+ private record ClusteringPushDownResult(Set<ColumnHandle> fullyPushedColumnPredicates, String domainQuery, boolean includesAllClusteringColumnsAndHasAllEqPredicates)`
`167`	`179`	`{`
`168`	`180`	`private ClusteringPushDownResult`
`169`	`181`	`{`
Original file line number	Diff line number	Diff line change
`@@ -34,23 +34,26 @@ public class CassandraNamedRelationHandle`
`34`	`34`	`private final String tableName;`
`35`	`35`	`private final Optional<List<CassandraPartition>> partitions;`
`36`	`36`	`private final String clusteringKeyPredicates;`
	`37`	`+ private final Boolean includesAllClusteringColumnsAndHasAllEqPredicates;`
`37`	`38`
`38`	`39`	`public CassandraNamedRelationHandle(String schemaName, String tableName)`
`39`	`40`	`{`
`40`		`- this(schemaName, tableName, Optional.empty(), "");`
	`41`	`+ this(schemaName, tableName, Optional.empty(), "", null);`
`41`	`42`	`}`
`42`	`43`
`43`	`44`	`@JsonCreator`
`44`	`45`	`public CassandraNamedRelationHandle(`
`45`	`46`	`@JsonProperty("schemaName") String schemaName,`
`46`	`47`	`@JsonProperty("tableName") String tableName,`
`47`	`48`	`@JsonProperty("partitions") Optional<List<CassandraPartition>> partitions,`
`48`		`- @JsonProperty("clusteringKeyPredicates") String clusteringKeyPredicates)`
	`49`	`+ @JsonProperty("clusteringKeyPredicates") String clusteringKeyPredicates,`
	`50`	`+ @JsonProperty("includesAllClusteringColumnsAndHasAllEqPredicates") Boolean includesAllClusteringColumnsAndHasAllEqPredicates)`
`49`	`51`	`{`
`50`	`52`	`this.schemaName = requireNonNull(schemaName, "schemaName is null");`
`51`	`53`	`this.tableName = requireNonNull(tableName, "tableName is null");`
`52`	`54`	`this.partitions = partitions.map(ImmutableList::copyOf);`
`53`	`55`	`this.clusteringKeyPredicates = requireNonNull(clusteringKeyPredicates, "clusteringKeyPredicates is null");`
	`56`	`+ this.includesAllClusteringColumnsAndHasAllEqPredicates = includesAllClusteringColumnsAndHasAllEqPredicates;`
`54`	`57`	`}`
`55`	`58`
`56`	`59`	`@JsonProperty`
`@@ -77,6 +80,12 @@ public String getClusteringKeyPredicates()`
`77`	`80`	`return clusteringKeyPredicates;`
`78`	`81`	`}`
`79`	`82`
	`83`	`+ @JsonProperty`
	`84`	`+ public Boolean getIncludesAllClusteringColumnsAndHasAllEqPredicates()`
	`85`	`+ {`
	`86`	`+ return includesAllClusteringColumnsAndHasAllEqPredicates;`
	`87`	`+ }`
	`88`	`+`
`80`	`89`	`public SchemaTableName getSchemaTableName()`
`81`	`90`	`{`
`82`	`91`	`return new SchemaTableName(schemaName, tableName);`
`@@ -101,7 +110,8 @@ public boolean equals(Object obj)`
`101`	`110`	`return Objects.equals(this.schemaName, other.schemaName) &&`
`102`	`111`	`Objects.equals(this.tableName, other.tableName) &&`
`103`	`112`	`Objects.equals(this.partitions, other.partitions) &&`
`104`		`- Objects.equals(this.clusteringKeyPredicates, other.clusteringKeyPredicates);`
	`113`	`+ Objects.equals(this.clusteringKeyPredicates, other.clusteringKeyPredicates) &&`
	`114`	`+ Objects.equals(this.includesAllClusteringColumnsAndHasAllEqPredicates, other.includesAllClusteringColumnsAndHasAllEqPredicates);`
`105`	`115`	`}`
`106`	`116`
`107`	`117`	`@Override`
`@@ -122,6 +132,9 @@ public String toString()`
`122`	`132`	`if (!clusteringKeyPredicates.isEmpty()) {`
`123`	`133`	`result += format(" constraint(%s)", clusteringKeyPredicates);`
`124`	`134`	`}`
	`135`	`+ if (includesAllClusteringColumnsAndHasAllEqPredicates != null) {`
	`136`	`+ result += format(" includesAllClusteringColumnsAndHasAllEqPredicates(%s)", includesAllClusteringColumnsAndHasAllEqPredicates);`
	`137`	`+ }`
`125`	`138`	`return result;`
`126`	`139`	`}`
`127`	`140`	`}`
Original file line number	Diff line number	Diff line change
`@@ -462,7 +462,7 @@ private static Map<String, Integer> indexColumns(List<ColumnHandle> columnHandle`
`462`	`462`	`private CassandraTableHandle getTableHandle(Optional<List<CassandraPartition>> partitions, String clusteringKeyPredicates)`
`463`	`463`	`{`
`464`	`464`	`CassandraNamedRelationHandle handle = ((CassandraTableHandle) getTableHandle(tableForDelete)).getRequiredNamedRelation();`
`465`		`- return new CassandraTableHandle(new CassandraNamedRelationHandle(handle.getSchemaName(), handle.getTableName(), partitions, clusteringKeyPredicates));`
	`465`	`+ return new CassandraTableHandle(new CassandraNamedRelationHandle(handle.getSchemaName(), handle.getTableName(), partitions, clusteringKeyPredicates, null));`
`466`	`466`	`}`
`467`	`467`
`468`	`468`	`private CassandraPartition createPartition(long value1, long value2)`