Fix CQLSSTableWriter serialization of vector of date and time

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Fix CQLSSTableWriter serialization of vector of date and time (CASSANDRA-20979)
  * Enable CQLSSTableWriter to create SSTables compressed with a dictionary (CASSANDRA-20938)
  * Support ZSTD dictionary compression (CASSANDRA-17021)
  * Fix ExceptionsTable when stacktrace has zero elements (CASSANDRA-20992)

src/java/org/apache/cassandra/db/marshal/SimpleDateType.java

@@ -124,4 +124,10 @@ public ByteBuffer getMaskedValue()
     {
         return MASKED_VALUE;
     }
+
+    @Override
+    public int valueLengthIfFixed()
+    {
+        return 4;
+    }
 }

src/java/org/apache/cassandra/db/marshal/TimeType.java

@@ -120,4 +120,10 @@ public ByteBuffer getMaskedValue()
     {
         return DEFAULT_MASKED_VALUE;
     }
+
+    @Override
+    public int valueLengthIfFixed()
+    {
+        return 8;
+    }
 }

test/unit/org/apache/cassandra/db/marshal/AbstractTypeTest.java

@@ -960,17 +960,31 @@ public void testAssumedCompatibility()
     @Test
     public void testBackwardCompatibility()
     {
+        // Fix for CASSANDRA-20979 altered behaviour of SimpleDateType and TimeType by correctly
+        // returning the fix-length size in valueLengthIfFixed(). This change impacted logic of
+        // isSerializationCompatibleWith(), which as a result complains about backward compatibility of:
+        // 1) [SimpleDateType isSerializationCompatibleWith Int32Type, {}] expected:<[fals]e> but was:<[tru]e>
+        // 2) [TimeType isSerializationCompatibleWith LongType, {}] expected:<[fals]e> but was:<[tru]e>
+        // 3) [BytesType isSerializationCompatibleWith SimpleDateType, {}] expected:<[tru]e> but was:<[fals]e>
+        // 4) [BytesType isSerializationCompatibleWith TimeType, {}] expected:<[tru]e> but was:<[fals]e>
+        ImmutableMap<AbstractType, Set<AbstractType>> exclusions = ImmutableMap.<AbstractType, Set<AbstractType>>builder()
+                                                                               .put(SimpleDateType.instance, ImmutableSet.of(Int32Type.instance))
+                                                                               .put(TimeType.instance, ImmutableSet.of(LongType.instance))
+                                                                               .put(BytesType.instance, ImmutableSet.of(SimpleDateType.instance, TimeType.instance))
+                                                                               .build();
+
         cassandra40TypesCompatibility.assertLoaded();
-        testBackwardCompatibility(currentTypesCompatibility, cassandra40TypesCompatibility);
+        testBackwardCompatibility(currentTypesCompatibility, cassandra40TypesCompatibility, exclusions);
 
         cassandra41TypesCompatibility.assertLoaded();
-        testBackwardCompatibility(currentTypesCompatibility, cassandra41TypesCompatibility);
+        testBackwardCompatibility(currentTypesCompatibility, cassandra41TypesCompatibility, exclusions);
 
         cassandra50TypesCompatibility.assertLoaded();
-        testBackwardCompatibility(currentTypesCompatibility, cassandra50TypesCompatibility);
+        testBackwardCompatibility(currentTypesCompatibility, cassandra50TypesCompatibility, exclusions);
     }
 
-    public void testBackwardCompatibility(TypesCompatibility upgradeTo, TypesCompatibility upgradeFrom)
+    public void testBackwardCompatibility(TypesCompatibility upgradeTo, TypesCompatibility upgradeFrom,
+                                          Map<AbstractType, Set<AbstractType>> exclusions)
     {
         SoftAssertions assertions = new SoftAssertionsWithLimit(100);
 
@@ -981,6 +995,8 @@ public void testBackwardCompatibility(TypesCompatibility upgradeTo, TypesCompati
         assertions.assertThat(upgradeTo.multiCellSupportingTypesForReading()).containsAll(upgradeFrom.multiCellSupportingTypes());
 
         forEachTypesPair(true, (l, r) -> {
+            if (skipTypeCheck(l, r, exclusions))
+                return;
             if (upgradeFrom.expectCompatibleWith(l, r))
                 assertions.assertThat(upgradeTo.expectCompatibleWith(l, r)).describedAs(isCompatibleWithDesc(l, r)).isTrue();
             if (upgradeFrom.expectSerializationCompatibleWith(l, r))
@@ -992,6 +1008,65 @@ public void testBackwardCompatibility(TypesCompatibility upgradeTo, TypesCompati
         assertions.assertAll();
     }
 
+    private boolean skipTypeCheck(AbstractType left, AbstractType right, Map<AbstractType, Set<AbstractType>> exclusions)
+    {
+        for (AbstractType excludeLeft : exclusions.keySet())
+        {
+            if (hasPrimitiveType(left, excludeLeft))
+            {
+                for (AbstractType excludeRight : exclusions.get(excludeLeft))
+                {
+                    if (hasPrimitiveType(right, excludeRight))
+                    {
+                        return true;
+                    }
+                }
+            }
+        }
+        return false;
+    }
+
+    private boolean hasPrimitiveType(AbstractType compositeType, AbstractType primitiveType)
+    {
+        if (compositeType == primitiveType)
+        {
+            return true;
+        }
+        else if (compositeType instanceof ListType)
+        {
+            ListType listType = (ListType) compositeType;
+            return hasPrimitiveType(listType.getElementsType(), primitiveType);
+        }
+        else if (compositeType instanceof SetType)
+        {
+            SetType setType = (SetType) compositeType;
+            return hasPrimitiveType(setType.getElementsType(), primitiveType);
+        }
+        else if (compositeType instanceof MapType)
+        {
+            MapType mapType = (MapType) compositeType;
+            return hasPrimitiveType(mapType.getKeysType(), primitiveType)
+                   || hasPrimitiveType(mapType.getValuesType(), primitiveType);
+        }
+        else if (compositeType instanceof VectorType)
+        {
+            VectorType vectorType = (VectorType) compositeType;
+            return hasPrimitiveType(vectorType.getElementsType(), primitiveType);
+        }
+        else if (compositeType instanceof TupleType)
+        {
+            TupleType tupleType = (TupleType) compositeType;
+            for (int i = 0; i < tupleType.size(); i++)
+            {
+                if (hasPrimitiveType(tupleType.type(i), primitiveType))
+                {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
     @Test
     public void testImplementedCompatibility()
     {
@@ -1593,6 +1668,7 @@ public void assertLoaded()
         {
             loadAssertions.assertAll();
         }
+
         @Override
         public boolean expectCompatibleWith(AbstractType left, AbstractType right)
         {
@@ -1694,11 +1770,11 @@ private CurrentTypesCompatibility()
             primitiveSerializationCompatibleWith.put(BytesType.instance, InetAddressType.instance);
             primitiveSerializationCompatibleWith.put(BytesType.instance, IntegerType.instance);
             primitiveSerializationCompatibleWith.put(BytesType.instance, ShortType.instance);
-            primitiveSerializationCompatibleWith.put(BytesType.instance, SimpleDateType.instance);
-            primitiveSerializationCompatibleWith.put(BytesType.instance, TimeType.instance);
             primitiveSerializationCompatibleWith.put(BytesType.instance, UTF8Type.instance);
             primitiveSerializationCompatibleWith.put(LongType.instance, TimestampType.instance);
+            primitiveSerializationCompatibleWith.put(SimpleDateType.instance, Int32Type.instance);
             primitiveSerializationCompatibleWith.put(TimestampType.instance, LongType.instance);
+            primitiveSerializationCompatibleWith.put(TimeType.instance, LongType.instance);
             primitiveSerializationCompatibleWith.put(UTF8Type.instance, AsciiType.instance);
             primitiveSerializationCompatibleWith.put(UUIDType.instance, TimeUUIDType.instance);
 

test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java

@@ -36,7 +36,9 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiConsumer;
 import java.util.function.BiPredicate;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
@@ -51,6 +53,7 @@
 
 import com.datastax.driver.core.utils.UUIDs;
 import org.apache.cassandra.Util;
+import org.apache.cassandra.cql3.CQL3Type;
 import org.apache.cassandra.cql3.QueryProcessor;
 import org.apache.cassandra.cql3.UntypedResultSet;
 import org.apache.cassandra.cql3.constraints.ConstraintViolationException;
@@ -64,7 +67,10 @@
 import org.apache.cassandra.db.compression.CompressionDictionaryTrainingConfig;
 import org.apache.cassandra.db.compression.ZstdCompressionDictionary;
 import org.apache.cassandra.db.compression.ZstdDictionaryTrainer;
+import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.db.marshal.FloatType;
+import org.apache.cassandra.db.marshal.SimpleDateType;
+import org.apache.cassandra.db.marshal.TimeType;
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.dht.ByteOrderedPartitioner;
 import org.apache.cassandra.dht.Murmur3Partitioner;
@@ -1623,9 +1629,36 @@ public void testSkipBuildingIndexesWithSAI() throws Exception
     @Test
     public void testWritingVectorData() throws Exception
     {
+        testWritingVectorData(CQL3Type.Native.FLOAT, FloatType.instance, (i) -> (float) i, (i, vector) -> {
+            assertThat(vector).allMatch(val -> val instanceof Float);
+            assertThat(vector).allMatch(val -> (float) val == (float) i);
+        });
+
+        perTestSetup();
+
+        testWritingVectorData(CQL3Type.Native.DATE, SimpleDateType.instance, LocalDate::fromDaysSinceEpoch, (i, vector) -> {
+            assertThat(vector).allMatch(val -> val instanceof Integer);
+            assertThat(vector).allMatch(val -> {
+                int days = (int) val - Integer.MIN_VALUE; // signed to unsigned conversion
+                return days == i;
+            });
+        });
+
+        perTestSetup();
+
+        testWritingVectorData(CQL3Type.Native.TIME, TimeType.instance, (i) -> (long) i, (i, vector) -> {
+            assertThat(vector).allMatch(val -> val instanceof Long);
+            assertThat(vector).allMatch(val -> (long) val == (long) i);
+        });
+    }
+
+    private void testWritingVectorData(CQL3Type.Native cqlType, AbstractType<?> subType, Function<Integer, ?> valueFactory,
+                                       BiConsumer<Integer, List<?>> checkFunction) throws Exception
+    {
+        final int dimensions = 5;
         final String schema = "CREATE TABLE " + qualifiedTable + " ("
                               + "  k int,"
-                              + "  v1 VECTOR<FLOAT, 5>,"
+                              + "  v1 VECTOR<" + cqlType.name() + ", " + dimensions + ">,"
                               + "  PRIMARY KEY (k)"
                               + ")";
 
@@ -1637,7 +1670,12 @@ public void testWritingVectorData() throws Exception
 
         for (int i = 0; i < 100; i++)
         {
-            writer.addRow(i, List.of( (float)i, (float)i, (float)i, (float)i, (float)i));
+            List<Object> vector = new ArrayList<>(dimensions);
+            for (int j = 0; j < dimensions; j++)
+            {
+                vector.add(valueFactory.apply(i));
+            }
+            writer.addRow(i, vector);
         }
 
         writer.close();
@@ -1652,10 +1690,9 @@ public void testWritingVectorData() throws Exception
             for (UntypedResultSet.Row row : resultSet)
             {
                 assertEquals(cnt, row.getInt("k"));
-                List<Float> vector = row.getVector("v1", FloatType.instance, 5);
-                assertThat(vector).hasSize(5);
-                final float floatCount = (float)cnt;
-                assertThat(vector).allMatch(val -> val == floatCount);
+                List<?> vector = row.getVector("v1", subType, dimensions);
+                assertThat(vector).hasSize(dimensions);
+                checkFunction.accept(cnt, vector);
                 cnt++;
             }
         }