Merge branch 'cassandra-6.0' into CASSANDRA-21134-direct-compaction-writes

Author: samueldlightfoot

debian/cassandra-tools.install

@@ -6,5 +6,5 @@ tools/bin/sstablemetadata usr/bin
 tools/bin/sstableofflinerelevel usr/bin
 tools/bin/sstablerepairedset usr/bin
 tools/bin/sstablesplit usr/bin
-tools/bin/addtocmstool usr/bin
+tools/bin/cmsofflinetool usr/bin
 tools/bin/offlineclustermetadatadump usr/bin

redhat/cassandra.spec

@@ -214,7 +214,7 @@ This package contains extra tools for working with Cassandra clusters.
 %attr(755,root,root) %{_bindir}/fqltool
 %attr(755,root,root) %{_bindir}/generatetokens
 %attr(755,root,root) %{_bindir}/hash_password
-%attr(755,root,root) %{_bindir}/addtocmstool
+%attr(755,root,root) %{_bindir}/cmsofflinetool
 %attr(755,root,root) %{_bindir}/offlineclustermetadatadump
 
 

src/java/org/apache/cassandra/cql3/Attributes.java

@@ -82,12 +82,12 @@ public boolean isTimeToLiveSet()
         return timeToLive != null;
     }
 
-    public long getTimestamp(long now, QueryOptions options) throws InvalidRequestException
+    public long getTimestamp(long now, FunctionContext context) throws InvalidRequestException
     {
         if (timestamp == null)
             return now;
 
-        ByteBuffer tval = timestamp.bindAndGet(options);
+        ByteBuffer tval = timestamp.bindAndGet(context);
         if (tval == null)
             throw new InvalidRequestException("Invalid null value of timestamp");
 
@@ -106,15 +106,15 @@ public long getTimestamp(long now, QueryOptions options) throws InvalidRequestEx
         return LongType.instance.compose(tval);
     }
 
-    public int getTimeToLive(QueryOptions options, TableMetadata metadata) throws InvalidRequestException
+    public int getTimeToLive(FunctionContext context, TableMetadata metadata) throws InvalidRequestException
     {
         if (timeToLive == null)
         {
             ExpirationDateOverflowHandling.maybeApplyExpirationDateOverflowPolicy(metadata, metadata.params.defaultTimeToLive, true);
             return metadata.params.defaultTimeToLive;
         }
 
-        ByteBuffer tval = timeToLive.bindAndGet(options);
+        ByteBuffer tval = timeToLive.bindAndGet(context);
         if (tval == null)
             return 0;
 

src/java/org/apache/cassandra/cql3/ColumnsExpression.java

@@ -356,9 +356,9 @@ public Kind kind()
      * Returns the key, index or fieldname specifying the selected element.
      * @return the key, index or fieldname specifying the selected element.
      */
-    public ByteBuffer element(QueryOptions options)
+    public ByteBuffer element(FunctionContext context)
     {
-        return element.bindAndGet(options);
+        return element.bindAndGet(context);
     }
 
     /**

src/java/org/apache/cassandra/cql3/ElementExpression.java

@@ -158,12 +158,12 @@ public void addFunctionsTo(List<Function> functions)
     /**
      * Returns the ByteBuffer representation of the key or index.
      *
-     * @param options the query options
+     * @param context the query options
      * @return the ByteBuffer representation of the key or index.
      */
-    public ByteBuffer bindAndGet(QueryOptions options)
+    public ByteBuffer bindAndGet(FunctionContext context)
     {
-        return keyOrIndex.bindAndGet(options);
+        return keyOrIndex.bindAndGet(context);
     }
 
     public String toCQLString()

src/java/org/apache/cassandra/cql3/FunctionContext.java

@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.cassandra.cql3;
+
+import java.time.Instant;
+
+import org.apache.cassandra.cql3.functions.Arguments;
+import org.apache.cassandra.cql3.functions.FunctionArguments;
+import org.apache.cassandra.db.marshal.TimeUUIDType;
+import org.apache.cassandra.transport.ProtocolVersion;
+import org.apache.cassandra.utils.Clock.Global;
+import org.apache.cassandra.utils.TimeUUID;
+
+public interface FunctionContext
+{
+    NoTimeOrQueryFunctionContext NONE = new NoTimeOrQueryFunctionContext() {};
+
+    interface RealTimeFunctionContext extends FunctionContext
+    {
+        @Override default byte[] nextTimeUUIDAsBytes() { return TimeUUID.Generator.nextTimeUUIDAsBytes(); }
+        @Override default Instant now() { return Global.currentTime(); }
+        @Override default long nowMicros() { return Global.currentTimeMicros(); }
+        @Override default long nowMillis() { return Global.currentTimeMillis(); }
+    }
+
+    interface NoTimeFunctionContext extends FunctionContext
+    {
+        @Override default byte[] nextTimeUUIDAsBytes() { throw new UnsupportedOperationException(); }
+        @Override default Instant now() { throw new UnsupportedOperationException(); }
+        @Override default long nowMicros() { throw new UnsupportedOperationException(); }
+        @Override default long nowMillis() { throw new UnsupportedOperationException(); }
+    }
+
+    interface NoTimeOrQueryFunctionContext extends NoTimeFunctionContext
+    {
+        @Override default QueryOptions options() { throw new UnsupportedOperationException(); }
+    }
+
+    interface PartialFunctionContext extends FunctionContext
+    {
+        default long nowMillis() { return nowMicros() / 1000; }
+        default Instant now()
+        {
+            long nowMicros = nowMicros();
+            return Instant.ofEpochSecond(nowMicros / 1000_000, (nowMicros % 1000_000) * 1000);
+        }
+    }
+
+    abstract class MicrosFunctionContext implements PartialFunctionContext
+    {
+        final long atMicros;
+        private long timeUuidNanos;
+
+        public MicrosFunctionContext(long atMicros)
+        {
+            this.atMicros = atMicros;
+        }
+
+        @Override public long nowMicros() { return atMicros; }
+
+        @Override
+        public byte[] nextTimeUUIDAsBytes()
+        {
+            return TimeUUID.toBytes(TimeUUID.unixMicrosToMsb(atMicros), TimeUUIDType.signedBytesToNativeLong(timeUuidNanos++));
+        }
+    }
+
+    QueryOptions options();
+    Instant now();
+    long nowMillis();
+    long nowMicros();
+    byte[] nextTimeUUIDAsBytes();
+
+    default ProtocolVersion getProtocolVersion() { return options().getProtocolVersion(); }
+
+    default Arguments noArguments()
+    {
+        return new FunctionArguments(this);
+    }
+}

src/java/org/apache/cassandra/cql3/Json.java

@@ -251,12 +251,12 @@ public boolean containsBindMarker()
         }
 
         @Override
-        public Terminal bind(QueryOptions options) throws InvalidRequestException
+        public Terminal bind(FunctionContext context) throws InvalidRequestException
         {
-            Term term = options.getJsonColumnValue(marker.bindIndex, column.name, marker.columns);
+            Term term = context.options().getJsonColumnValue(marker.bindIndex, column.name, marker.columns);
             return term == null
                    ? (defaultUnset ? Constants.UNSET_VALUE : null)
-                   : term.bind(options);
+                   : term.bind(context);
         }
 
         @Override

src/java/org/apache/cassandra/cql3/Operation.java

@@ -116,9 +116,9 @@ public void collectMarkerSpecification(VariableSpecifications boundNames, Object
      * Execute the operation.
      *
      * @param partitionKey partition key for the update.
-     * @param params parameters of the update.
+     * @param builder      parameters of the update.
      */
-    public abstract void execute(DecoratedKey partitionKey, UpdateParameters params) throws InvalidRequestException;
+    public abstract void execute(DecoratedKey partitionKey, RowUpdateBuilder builder) throws InvalidRequestException;
 
     /**
      * A parsed raw UPDATE operation.

src/java/org/apache/cassandra/cql3/QueryOptions.java

@@ -31,6 +31,7 @@
 
 import org.apache.cassandra.config.DataStorageSpec;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.FunctionContext.RealTimeFunctionContext;
 import org.apache.cassandra.cql3.terms.Term;
 import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.db.marshal.UTF8Type;
@@ -53,7 +54,7 @@
 /**
  * Options for a query.
  */
-public abstract class QueryOptions
+public abstract class QueryOptions implements RealTimeFunctionContext
 {
     public static final QueryOptions DEFAULT = new DefaultQueryOptions(ConsistencyLevel.ONE,
                                                                        Collections.emptyList(),
@@ -378,6 +379,12 @@ public long getCoordinatorReadSizeFailThresholdBytes()
         }
     }
 
+    @Override
+    public QueryOptions options()
+    {
+        return this;
+    }
+
     static class DefaultQueryOptions extends QueryOptions
     {
         private final ConsistencyLevel consistency;

src/java/org/apache/cassandra/cql3/QueryProcessor.java

@@ -555,7 +555,7 @@ else if (readQuery instanceof SinglePartitionReadQuery.Group)
                                                                                       .map(m -> MessagingService.instance().<ReadCommand, ReadResponse>sendWithResult(m, address))
                                                                                       .collect(Collectors.toList()));
 
-            ResultSetBuilder result = new ResultSetBuilder(select.getResultMetadata(), select.getSelection().newSelectors(options), false);
+            ResultSetBuilder result = new ResultSetBuilder(select.getResultMetadata(), options, select.getSelection().newSelectors(options), false);
             return future.map(list -> {
                 int i = 0;
                 for (Message<ReadResponse> m : list)