Fix Cassandra Range pushdown #8629

s2lomon · 2021-07-21T12:29:08Z

It's a pr that tries to fix this issue #401. First by fixing the correctness (so disabling the pushdown all together).

It seems to me however that logic which drives ranges pushdown in CassandraClusteringPredicatesExtractor is trying to deal with generic ranges and joint them with AND operator, when in reality these ranges comes from SortedRangeSet and should be treat as a sum of Range rather than an intersection. Since Cassandra doesn't support OR operator, I think that we should only push single range values all the time - so in a simplest scenario we should only handle pushdown for SortedRangeSets that size equals 1.

I would like to quickly refactor CassandraClusteringPredicatesExtractor's range handling to reflect that directly.

Ok I've clearly forgotten about IN statement, which results in similar situation. I would stil refactor this though, just with a bit more subtle approach.

findepi · 2021-07-21T20:34:24Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

+                        if (rangesAreEquivalentToSingleValueNegation(orderedRanges)) {
+                            return null;
+                        }
+                        for (Range range : orderedRanges) {


the predicate = Joiner.on(" AND ").join(rangeConjuncts); below is incorrect, isn't it?
Should be " OR " i guess?

is it what the bug report is actually about?

Yes, the pointed code is the cause. However, Cassandra doesn't support OR.

so we should remove that code -- first.
and limit pushdowns to single-value (or INs).

and this will implicitly solve the negation problem as well

Ok so this would fix the issue, but we would lose pushdown of a single multivalued ranges, which is not that great.
I think that we need to detect when we should skip pushing down

Ok so this would fix the issue, but we would lose pushdown of a single multivalued ranges, which is not that great.

i didn't say that

depends how the code is written

findepi · 2021-07-21T20:36:52Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

+        return first.intersect(second).isEmpty()
+                && !first.isHighInclusive()
+                && !first.isHighUnbounded()
+                && !second.isLowInclusive()
+                && !second.isLowUnbounded()
+                && first.getHighBoundedValue().equals(second.getLowBoundedValue());


Is it equivalent to return domain.complement().isSingleValue()?

Yes that's what it should detect. I have a different solution however.

plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestCassandraConnectorTest.java

findepi · 2021-07-23T09:44:58Z

Cassandra seems to support

=, <, >, ....
IN (....)
... AND ...

So:

When we have single single-valued range, we should use =.
When we have single range, we should use low bound < x AND x < high bound (or <= when appropriate)
When we have multiple single-valued range, we should use IN (...).
In all other cases we should push down min/max bounds (domain.getValues().getRanges().getSpan()) using low bound < x AND x < high bound (or <= when appropriate)

BTW can you please give the PR appropriate title after we found our the root cause?

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

findepi · 2021-07-26T13:56:02Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

+
+                            return discreteValues.getValues().stream()
+                                    .map(columnHandle.getCassandraType()::toCqlLiteral)
+                                    .reduce((first, second) -> first + "," + second)


is this O(n²) ?

use .collect( [Collectors.] joining(",", "IN(", ")") )

Good catch, changing it to collect anyway.

findepi · 2021-07-26T13:56:58Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

+    }
+
+    /**
+     * IN restriction allowed only on last clustering column for Cassandra version = 2.1


Code seems to indicate that. I've just moved it around. Will change the comment.

findepi · 2021-07-26T13:57:40Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

+        return cassandraVersion.compareTo(VersionNumber.parse("2.2.0")) < 0 && currentlyProcessedClusteringColumn != (clusteringColumns.size() - 1);
+    }
+
+    private static String translateRangeIntoCQL(CassandraColumnHandle columnHandle, Range range)


findepi · 2021-07-26T13:58:47Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

    }

    private static ClusteringPushDownResult getClusteringKeysSet(List<CassandraColumnHandle> clusteringColumns, TupleDomain<ColumnHandle> predicates, VersionNumber cassandraVersion)
    {
-        ImmutableMap.Builder<ColumnHandle, Domain> domainsBuilder = ImmutableMap.builder();
+        ImmutableMap.Builder<ColumnHandle, Domain> fullyPushedDomains = ImmutableMap.builder();


fullyPushedDomains is a Map, but its values are apparently never read. did you mean to make it a Set?

So I've left it as a map, as that's how it has been implemented before (although it's values weren't used), I've just changed the names a bit to better reflect what's going on here. It could be a Set though. I can change it to one.

So I've left it as a map, as that's how it has been implemented before (although it's values weren't used)

could be nice prep commit

I've just changed the names a bit to better reflect what's going on here

that's a good change in general

as it is a Map, i though we're comparing fully pushed down domains with some other domains, so was a bit confused when realized this isn't the case

findepi

editorials

over to @hashhar

findepi · 2021-07-27T13:24:49Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

-                            predicate = Joiner.on(" AND ").join(rangeConjuncts);
+                        if (ranges.getOrderedRanges().stream().allMatch(Range::isSingleValue)) {
+                            if (isInStatementNotAllowed(clusteringColumns, cassandraVersion, currentlyProcessedClusteringColumn)) {
+                                return null;


You could still do return translateRangeIntoCql(columnHandle, ranges.getSpan()); (as below)

You are right, let's do that.

I will try to do the same for the discreteValues - as we are not pushing at all in case of IN beeing not supported.

findepi · 2021-07-27T13:28:10Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

-                        }
-
-                        if (!singleValues.isEmpty() && !rangeConjuncts.isEmpty()) {
+                        if (ranges.getSpan().isAll()) {
                            return null;


if (ranges.getSpan().isAll()) then you still could benefit from ranges.getOrderedRanges().stream().allMatch(Range::isSingleValue) handling.

(we don't have a logic like that in core yet, but technically {10, 5, MIN_VALUE, MAX_VALUE} is set of 4 values, with span being "all")

It seems that you can simply remove

if (ranges.getSpan().isAll()) { return null; }

block since, this case nicely handled by the following code anyway

Hm we still need to do this check, to not be pushing WHERE x > MIN AND x < MAX, as it doesn't make sense. I can move it to the method responsible for handling range pushdown though.

findepi · 2021-07-27T13:29:08Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

-                            }
+                        if (ranges.getRangeCount() == 1) {
+                            fullyPushedDomains.add(columnHandle);
+                            return translateRangeIntoCql(columnHandle, ranges.getOrderedRanges().get(0));


nit: .get(0) -> [Iterables.] getOnlyElement
(copy&paste fool-proof)

findepi · 2021-07-27T13:30:12Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

+                                        discreteValues.getValues().stream().findFirst()
+                                                .orElseThrow());


getOnlyElement(discreteValues.getValues())

findepi · 2021-07-27T13:33:45Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

+    /**
+     * IN restriction allowed only on last clustering column for Cassandra version <= 2.1
+     */
+    private static boolean isInStatementNotAllowed(List<CassandraColumnHandle> clusteringColumns, VersionNumber cassandraVersion, int currentlyProcessedClusteringColumn)


isIn...Allowed (without "Not") would read better.

intellij's suggestion Java | Data flow | Boolean method is always inverted is not something i would follow unconditionally

also "InStatement" -> "InExpression"

It's just to not force a reader to go through !true -> not isInStatementAllowed, which is less descriptive than full sentence isInStatementNotAllowed. The other argument for leaving it as it is, is that we are only interested in the negative scenario.

findepi · 2021-07-27T13:34:26Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

        }

-        public Map<ColumnHandle, Domain> getDomains()
+        public boolean hasNotBeenFullyPushed(ColumnHandle column)


i would rather let the calling code to apply negation, so that method semantics are clearer.

Sure, makes sense as thanks to that this method is more descriptive in context of its class.

findepi · 2021-07-29T10:44:05Z

@s2lomon see CI results
i will leave this up to @hashhar for final merge.

hashhar

Some general comments. I'll defer to someone else like @ebyhr or @raunaqmorarka for the correctness of the impl.

hashhar · 2021-07-30T06:59:00Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

    }

    private static ClusteringPushDownResult getClusteringKeysSet(List<CassandraColumnHandle> clusteringColumns, TupleDomain<ColumnHandle> predicates, VersionNumber cassandraVersion)
    {
-        ImmutableMap.Builder<ColumnHandle, Domain> domainsBuilder = ImmutableMap.builder();
+        ImmutableSet.Builder<ColumnHandle> fullyPushedDomains = ImmutableSet.builder();


nit: Extract the Map -> Set changes to a separate commit. Makes it easy to differentiate the fix from other changes.

hashhar · 2021-07-30T07:01:29Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

+    }
+
+    /**
+     * IN restriction allowed only on last clustering column for Cassandra version <= 2.1


nit: Adjust the comment since the condition checks for v < 2.2.0 (there are versions between 2.1 and 2.2.0).

yep you are right

hashhar · 2021-07-30T07:05:35Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

+                        }
+                        if (ranges.getOrderedRanges().stream().allMatch(Range::isSingleValue)) {
+                            if (isInStatementNotAllowed(clusteringColumns, cassandraVersion, currentlyProcessedClusteringColumn)) {
+                                return translateRangeIntoCql(columnHandle, ranges.getSpan());


Some of the cases in translateRangeIntoCql perform full pushdown so should the column handle be added to fullyPushedDomains too?

So the idea here is to have a single method that deals with the complexity of creating a Cassandra predicate to pushdown for Ranges, and the general flow that recognizes when the pushdown is full or only partial. Here it's partial, as for most common scenario we are changing WHERE x IN (1,3,4) into WHERE x >=1 AND x<=4 - which will still include x=2 that needs to be filtered out.

plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestCassandraConnectorTest.java

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

raunaqmorarka · 2021-07-30T08:04:47Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

+                                    List<Range> ranges = discreteValues.getValues().stream()
+                                            .map(value -> Range.equal(domain.getType(), value))
+                                            .collect(toImmutableList());
+                                    return translateRangeIntoCql(columnHandle, SortedRangeSet.copyOf(domain.getType(), ranges).getSpan());


We get here for EquatableValueSet and that is used only for types are comparable but not orderable, so taking a span of these values is probably wrong. E.g. you could try this for ColorType, Range will probably fail to create comparisonOperator from that.

Ok so I should probably have a check here for isOrderable rather than isComparable and this pushdown should be fine, isn't it?

You will never get here when isOrderable is true because EquatableValueSet is used only when isOrderable is false and isComparable is true.

raunaqmorarka · 2021-07-30T08:15:14Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

@@ -65,102 +67,135 @@ private static ClusteringPushDownResult getClusteringKeysSet(List<CassandraColum
            if (domain.isNullAllowed()) {
                break;
            }
+
+            int currentlyProcessedClusteringColumn = allProcesedClusteringColumns;
            String predicateString = domain.getValues().getValuesProcessor().transform(


Can we just use something like

if (domain.isSingleValue()) { // construct equality predicate from domain.getSingleValue() } else if (domain.isNullableDiscreteSet()) { // construct IN predicate from domain.getNullableDiscreteSet() } else if (type.isOrderable()) { // construct range predicate from domain.getValues().getRanges().getSpan() }

In first 2 cases it is full pushdown, in 3rd case it is not

It might be equivalent but I'm not sure. If it is, let's make it a subsequent pr as it strides even further from the original implementation

I think this would be a safer and easier to understand approach but I'll leave it to @hashhar to decide if we should do it here or in a follow up.

That looks like a refactor and I agree it looks much simpler than the current code but let's keep it for a follow-up since the current code is a smaller diff that fixes the original issue.
Although I also don't know if it's logically equivalent or not.

hashhar · 2021-08-04T05:53:04Z

Let's also do #8629 (comment) in a follow-up since it helps a lot with making sense of what's going on.

ebyhr

Left only minor comments.

ebyhr · 2021-08-04T13:43:04Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

        ImmutableList.Builder<String> clusteringColumnSql = ImmutableList.builder();
-        int currentClusteringColumn = 0;
+        int allProcesedClusteringColumns = 0;


Fix typo: allProcessedClusteringColumns

ebyhr · 2021-08-04T14:20:08Z

...assandra/src/main/java/io/trino/plugin/cassandra/CassandraClusteringPredicatesExtractor.java

+                                    .map(range -> toCqlLiteral(columnHandle, range.getSingleValue()))
+                                    .collect(joining(","));
+                            fullyPushedColumnPredicates.add(columnHandle);
+                            return CassandraCqlUtils.validColumnName(columnHandle.getName()) + "IN (" + inValues + ")";


nit: Add a space before IN to avoid "column_name"IN (...

Due to the fact that Cassandra supports only =, <, >, .... IN (....) ... AND ... When we have single single-valued range, we use =. When we have single range, we use low bound < x AND x < high bound (or <= when appropriate) When we have multiple single-valued range, we use IN (...). In all other cases, including when IN is not supported in Cassandra, we push down min/max bounds (domain.getValues().getRanges().getSpan()) using low bound < x AND x < high bound (or <= when appropriate)

cla-bot bot added the cla-signed label Jul 21, 2021

s2lomon requested review from wendigo, findepi and hashhar July 21, 2021 12:29

hashhar requested a review from ebyhr July 21, 2021 12:42

findepi reviewed Jul 21, 2021

View reviewed changes

s2lomon force-pushed the fix/cassandra-clustering-key-negation branch from 5d1211b to e9b3947 Compare July 23, 2021 09:29

s2lomon changed the title ~~Disable Cassandra negation pushdown~~ Fix Cassandra Range pushdown Jul 26, 2021

s2lomon force-pushed the fix/cassandra-clustering-key-negation branch 2 times, most recently from cc1870e to bfd46c8 Compare July 26, 2021 12:34

findepi reviewed Jul 26, 2021

View reviewed changes

s2lomon force-pushed the fix/cassandra-clustering-key-negation branch from bfd46c8 to 925ecd8 Compare July 27, 2021 09:20

s2lomon requested a review from findepi July 27, 2021 12:15

findepi reviewed Jul 27, 2021

View reviewed changes

findepi assigned hashhar Jul 27, 2021

s2lomon force-pushed the fix/cassandra-clustering-key-negation branch from 925ecd8 to 5c90b90 Compare July 29, 2021 09:09

s2lomon requested a review from findepi July 29, 2021 09:10

findepi removed their request for review July 29, 2021 10:44

hashhar reviewed Jul 30, 2021

View reviewed changes

raunaqmorarka reviewed Jul 30, 2021

View reviewed changes

findepi force-pushed the master branch from 8538e49 to 1f896ea Compare July 30, 2021 22:13

Refactor pushdown markings in CassandraClusteringPredicatesExtractor

208cf92

s2lomon force-pushed the fix/cassandra-clustering-key-negation branch from 5c90b90 to 587f848 Compare August 3, 2021 12:10

s2lomon requested review from raunaqmorarka and hashhar August 3, 2021 12:12

raunaqmorarka approved these changes Aug 4, 2021

View reviewed changes

hashhar approved these changes Aug 4, 2021

View reviewed changes

ebyhr approved these changes Aug 4, 2021

View reviewed changes

s2lomon force-pushed the fix/cassandra-clustering-key-negation branch from 587f848 to 0138644 Compare August 9, 2021 08:39

s2lomon force-pushed the fix/cassandra-clustering-key-negation branch from 0138644 to 33ef97b Compare August 9, 2021 09:25

hashhar approved these changes Aug 9, 2021

View reviewed changes

hashhar added this to the 361 milestone Aug 9, 2021

hashhar merged commit 750000c into trinodb:master Aug 9, 2021

This was referenced Aug 10, 2021

Inequality on clustering key returns empty results in cassandra #401

Closed

Release notes for 361 #8732

Closed

		discreteValues.getValues().stream().findFirst()
		.orElseThrow());

Fix Cassandra Range pushdown #8629

Fix Cassandra Range pushdown #8629

Conversation

s2lomon commented Jul 21, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

findepi commented Jul 23, 2021

Choose a reason for hiding this comment

s2lomon Jul 26, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

findepi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

s2lomon Jul 28, 2021 • edited Loading

Choose a reason for hiding this comment

findepi commented Jul 29, 2021

hashhar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hashhar Aug 3, 2021 • edited Loading

Choose a reason for hiding this comment

hashhar commented Aug 4, 2021

ebyhr left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

s2lomon commented Jul 21, 2021 •

edited

Loading

s2lomon Jul 26, 2021 •

edited

Loading

s2lomon Jul 28, 2021 •

edited

Loading

hashhar Aug 3, 2021 •

edited

Loading