diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/reader/RowData.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/reader/RowData.java index b77295f50..f4f2f4243 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/reader/RowData.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/reader/RowData.java @@ -24,6 +24,7 @@ import com.google.common.annotations.VisibleForTesting; +import org.apache.cassandra.spark.data.CqlField; import org.apache.cassandra.spark.utils.ByteBufferUtils; /** @@ -35,6 +36,7 @@ public class RowData private ByteBuffer columnName; private ByteBuffer value; private long timestamp; + private int ttl; private BigInteger token; @VisibleForTesting boolean isNewPartition = false; @@ -49,6 +51,7 @@ public void setPartitionKeyCopy(ByteBuffer partitionKeyBytes, BigInteger token) this.value = null; this.isNewPartition = true; this.timestamp = 0L; + this.ttl = CqlField.NO_TTL; } public boolean isNewPartition() @@ -107,6 +110,18 @@ public long getTimestamp() return timestamp; } + // TTL + + public int getTtl() + { + return ttl; + } + + public void setTtl(int ttl) + { + this.ttl = ttl; + } + @Override public String toString() { diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/sparksql/Cell.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/sparksql/Cell.java index c22beb5f3..e75235f22 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/sparksql/Cell.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/sparksql/Cell.java @@ -23,14 +23,18 @@ public class Cell { public final Object[] values; public final int position; + public final boolean isPkCkOnly; public final boolean isNewRow; public final long timestamp; + public final int ttl; - Cell(Object[] values, int position, boolean isNewRow, long timestamp) + Cell(Object[] values, int position, boolean isPkCkOnly, boolean isNewRow, long timestamp, int ttl) { this.values = values; this.position = position; + this.isPkCkOnly = isPkCkOnly; this.isNewRow = isNewRow; this.timestamp = timestamp; + this.ttl = ttl; } } diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/sparksql/CellIterator.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/sparksql/CellIterator.java index 59707a6a1..5982dae8a 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/sparksql/CellIterator.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/sparksql/CellIterator.java @@ -211,7 +211,7 @@ private boolean getNext() throws IOException // columns are projected. The column we find is irrelevant because if we fall under this // condition it means that we are in a situation where the row has only PK + CK, but no // regular columns. - next = new Cell(values, firstProjectedValueColumnPositionOrZero, newRow, rowData.getTimestamp()); + next = new Cell(values, firstProjectedValueColumnPositionOrZero, true, newRow, rowData.getTimestamp(), rowData.getTtl()); return true; } @@ -235,7 +235,7 @@ private boolean getNext() throws IOException } // Update next Cell - next = new Cell(values, field.position(), newRow, rowData.getTimestamp()); + next = new Cell(values, field.position(), false, newRow, rowData.getTimestamp(), rowData.getTtl()); return true; } diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/MapUtils.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/MapUtils.java index 81c5e2673..5688a6dc0 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/MapUtils.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/MapUtils.java @@ -25,6 +25,7 @@ import java.util.Optional; import java.util.function.Function; import java.util.function.Supplier; +import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -249,6 +250,27 @@ public static String getOrDefault(Map options, String key, Strin return options.getOrDefault(lowerCaseKey(key), defaultValue); } + /** + * Returns sub-map with keys that match given prefix from original map. Prefix match is case-insensitive. + * + * @param options source map + * @param keyPrefix prefix of keys to be returned + * @param truncatePrefix whether to truncate prefix from output's map keys + * @param defaultValue default value returned when source map does not contain any key with given prefix + * @return sub-map with keys that match given prefix + */ + public static Map getKeysWithPrefix(Map options, String keyPrefix, + boolean truncatePrefix, Map defaultValue) + { + int prefixLength = keyPrefix.length(); + String lowerCasePrefix = lowerCaseKey(keyPrefix); + Map subMap = options.entrySet().stream() + .filter(entry -> lowerCaseKey(entry.getKey()).startsWith(lowerCasePrefix)) + .collect(Collectors.toMap(k -> truncatePrefix ? k.getKey().substring(prefixLength) : k.getKey(), + Map.Entry::getValue)); + return subMap.isEmpty() ? defaultValue : subMap; + } + /** * Method to check if key is present in {@code options} map. * diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java index ff7e0b67d..022a5a68a 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java @@ -88,8 +88,6 @@ import org.apache.cassandra.spark.data.partitioner.ConsistencyLevel; import org.apache.cassandra.spark.data.partitioner.Partitioner; import org.apache.cassandra.spark.data.partitioner.TokenPartitioner; -import org.apache.cassandra.spark.sparksql.LastModifiedTimestampDecorator; -import org.apache.cassandra.spark.sparksql.RowBuilder; import org.apache.cassandra.spark.sparksql.filters.SSTableTimeRangeFilter; import org.apache.cassandra.spark.utils.CqlUtils; import org.apache.cassandra.spark.utils.ReaderTimeProvider; @@ -101,12 +99,13 @@ import org.apache.cassandra.spark.validation.StartupValidatable; import org.apache.cassandra.spark.validation.StartupValidator; import org.apache.spark.sql.SparkSession; -import org.apache.spark.sql.catalyst.InternalRow; -import org.apache.spark.sql.types.DataType; import org.apache.spark.util.ShutdownHookManager; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; +import static org.apache.cassandra.spark.data.SchemaFeatureCustomizer.addCellLastModifiedTimestamp; +import static org.apache.cassandra.spark.data.SchemaFeatureCustomizer.addCellTtl; +import static org.apache.cassandra.spark.data.SchemaFeatureCustomizer.aliasLastModifiedTimestamp; import static org.apache.cassandra.spark.utils.CqlUtils.isTimeRangeFilterSupported; import static org.apache.cassandra.spark.utils.Properties.NODE_STATUS_NOT_CONSIDERED; @@ -142,7 +141,11 @@ public class CassandraDataLayer extends PartitionedDataLayer implements StartupV protected List requestedFeatures; protected Map rfMap; @Nullable - protected String lastModifiedTimestampField; + protected String rowLastModifiedTimestampField; + @Nullable + protected Map cellLastModifiedTimestampFields; + @Nullable + protected Map cellTtlFields; protected Set sstableVersionsOnCluster; // volatile in order to publish the reference for visibility protected volatile CqlTable cqlTable; @@ -172,7 +175,9 @@ public CassandraDataLayer(@NotNull ClientConfig options, this.enableStats = options.enableStats(); this.readIndexOffset = options.readIndexOffset(); this.useIncrementalRepair = options.useIncrementalRepair(); - this.lastModifiedTimestampField = options.lastModifiedTimestampField(); + this.rowLastModifiedTimestampField = options.rowLastModifiedTimestampField(); + this.cellLastModifiedTimestampFields = options.cellLastModifiedTimestampFields(); + this.cellTtlFields = options.cellTtlFields(); this.requestedFeatures = options.requestedFeatures(); this.sstableTimeRangeFilter = options.sstableTimeRangeFilter; } @@ -198,7 +203,9 @@ protected CassandraDataLayer(@Nullable String keyspace, boolean enableStats, boolean readIndexOffset, boolean useIncrementalRepair, - @Nullable String lastModifiedTimestampField, + @Nullable String rowLastModifiedTimestampField, + @Nullable Map cellLastModifiedTimestampFields, + @Nullable Map cellTtlFields, List requestedFeatures, @NotNull Map rfMap, TimeProvider timeProvider, @@ -221,11 +228,21 @@ protected CassandraDataLayer(@Nullable String keyspace, this.enableStats = enableStats; this.readIndexOffset = readIndexOffset; this.useIncrementalRepair = useIncrementalRepair; - this.lastModifiedTimestampField = lastModifiedTimestampField; this.requestedFeatures = requestedFeatures; - if (lastModifiedTimestampField != null) + this.rowLastModifiedTimestampField = rowLastModifiedTimestampField; + if (rowLastModifiedTimestampField != null) + { + aliasLastModifiedTimestamp(this.requestedFeatures, this.rowLastModifiedTimestampField); + } + this.cellLastModifiedTimestampFields = cellLastModifiedTimestampFields; + if (cellLastModifiedTimestampFields != null) + { + addCellLastModifiedTimestamp(this.requestedFeatures, this.cellLastModifiedTimestampFields); + } + this.cellTtlFields = cellTtlFields; + if (cellTtlFields != null) { - aliasLastModifiedTimestamp(this.requestedFeatures, this.lastModifiedTimestampField); + addCellTtl(this.requestedFeatures, this.cellTtlFields); } this.rfMap = rfMap; this.timeProvider = timeProvider; @@ -847,7 +864,9 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE this.enableStats = in.readBoolean(); this.readIndexOffset = in.readBoolean(); this.useIncrementalRepair = in.readBoolean(); - this.lastModifiedTimestampField = readNullable(in); + this.rowLastModifiedTimestampField = readNullable(in); + this.cellLastModifiedTimestampFields = readNullableObject(in); + this.cellTtlFields = readNullableObject(in); int features = in.readShort(); List requestedFeatures = new ArrayList<>(features); for (int feature = 0; feature < features; feature++) @@ -856,11 +875,13 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE requestedFeatures.add(SchemaFeatureSet.valueOf(featureName.toUpperCase())); } this.requestedFeatures = requestedFeatures; - // Has alias for last modified timestamp - if (this.lastModifiedTimestampField != null) + // initialize features + if (this.rowLastModifiedTimestampField != null) { - aliasLastModifiedTimestamp(this.requestedFeatures, this.lastModifiedTimestampField); + aliasLastModifiedTimestamp(this.requestedFeatures, this.rowLastModifiedTimestampField); } + addCellLastModifiedTimestamp(this.requestedFeatures, this.cellLastModifiedTimestampFields); + addCellTtl(this.requestedFeatures, this.cellTtlFields); this.rfMap = (Map) in.readObject(); this.timeProvider = new ReaderTimeProvider(in.readLong()); this.sstableTimeRangeFilter = (SSTableTimeRangeFilter) in.readObject(); @@ -901,12 +922,19 @@ private void writeObject(ObjectOutputStream out) throws IOException, ClassNotFou out.writeBoolean(this.readIndexOffset); out.writeBoolean(this.useIncrementalRepair); // If lastModifiedTimestampField exist, it aliases the LMT field - writeNullable(out, this.lastModifiedTimestampField); + writeNullable(out, this.rowLastModifiedTimestampField); + writeNullableObject(out, this.cellLastModifiedTimestampFields); + writeNullableObject(out, this.cellTtlFields); + // Serialize only distinct request features + List featureNames = requestedFeatures.stream() + .map(SchemaFeature::optionName) + .distinct() + .collect(Collectors.toList()); // Write the list of requested features: first write the size, then write the feature names - out.writeShort(this.requestedFeatures.size()); - for (SchemaFeature feature : requestedFeatures) + out.writeShort(featureNames.size()); + for (String featureName : featureNames) { - out.writeUTF(feature.optionName()); + out.writeUTF(featureName); } out.writeObject(this.rfMap); out.writeLong(timeProvider.referenceEpochInSeconds()); @@ -927,6 +955,19 @@ private static void writeNullable(ObjectOutputStream out, @Nullable String strin } } + private static void writeNullableObject(ObjectOutputStream out, @Nullable Object object) throws IOException + { + if (object == null) + { + out.writeBoolean(false); + } + else + { + out.writeBoolean(true); + out.writeObject(object); + } + } + @Nullable private static String readNullable(ObjectInputStream in) throws IOException { @@ -937,6 +978,16 @@ private static String readNullable(ObjectInputStream in) throws IOException return null; } + @Nullable + private static T readNullableObject(ObjectInputStream in) throws IOException, ClassNotFoundException + { + if (in.readBoolean()) + { + return (T) in.readObject(); + } + return null; + } + /** * Validates that all SSTables being read have versions that were observed in gossip info. * This catches cases where SSTables have unexpected versions that weren't seen during driver initialization. @@ -1039,11 +1090,23 @@ public void write(Kryo kryo, Output out, CassandraDataLayer dataLayer) out.writeBoolean(dataLayer.readIndexOffset); out.writeBoolean(dataLayer.useIncrementalRepair); // If lastModifiedTimestampField exist, it aliases the LMT field - out.writeString(dataLayer.lastModifiedTimestampField); + out.writeString(dataLayer.rowLastModifiedTimestampField); + out.writeBoolean(dataLayer.cellLastModifiedTimestampFields != null); + if (dataLayer.cellLastModifiedTimestampFields != null) + { + kryo.writeObject(out, dataLayer.cellLastModifiedTimestampFields); + } + out.writeBoolean(dataLayer.cellTtlFields != null); + if (dataLayer.cellTtlFields != null) + { + kryo.writeObject(out, dataLayer.cellTtlFields); + } // Write the list of requested features: first write the size, then write the feature names SchemaFeaturesListWrapper listWrapper = new SchemaFeaturesListWrapper(); + // Serialize only distinct request features listWrapper.requestedFeatureNames = dataLayer.requestedFeatures.stream() .map(SchemaFeature::optionName) + .distinct() .collect(Collectors.toList()); kryo.writeObject(out, listWrapper); kryo.writeObject(out, dataLayer.rfMap); @@ -1089,6 +1152,8 @@ public CassandraDataLayer read(Kryo kryo, Input in, Class ty in.readBoolean(), in.readBoolean(), in.readString(), + in.readBoolean() ? kryo.readObject(in, HashMap.class) : null, + in.readBoolean() ? kryo.readObject(in, HashMap.class) : null, kryo.readObject(in, SchemaFeaturesListWrapper.class).toList(), kryo.readObject(in, HashMap.class), new ReaderTimeProvider(in.readLong()), @@ -1204,45 +1269,4 @@ protected void await(CountDownLatch latch) throw new RuntimeException(exception); } } - - static void aliasLastModifiedTimestamp(List requestedFeatures, String alias) - { - SchemaFeature featureAlias = new SchemaFeature() - { - @Override - public String optionName() - { - return SchemaFeatureSet.LAST_MODIFIED_TIMESTAMP.optionName(); - } - - @Override - public String fieldName() - { - return alias; - } - - @Override - public DataType fieldDataType() - { - return SchemaFeatureSet.LAST_MODIFIED_TIMESTAMP.fieldDataType(); - } - - @Override - public RowBuilder decorate(RowBuilder builder) - { - return new LastModifiedTimestampDecorator<>(builder, alias); - } - - @Override - public boolean fieldNullable() - { - return SchemaFeatureSet.LAST_MODIFIED_TIMESTAMP.fieldNullable(); - } - }; - int index = requestedFeatures.indexOf(SchemaFeatureSet.LAST_MODIFIED_TIMESTAMP); - if (index >= 0) - { - requestedFeatures.set(index, featureAlias); - } - } } diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java index 4ca6c9214..d0d0b4e77 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/ClientConfig.java @@ -38,7 +38,9 @@ import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; -import static org.apache.cassandra.spark.data.CassandraDataLayer.aliasLastModifiedTimestamp; +import static org.apache.cassandra.spark.data.SchemaFeatureCustomizer.addCellLastModifiedTimestamp; +import static org.apache.cassandra.spark.data.SchemaFeatureCustomizer.addCellTtl; +import static org.apache.cassandra.spark.data.SchemaFeatureCustomizer.aliasLastModifiedTimestamp; import static org.apache.cassandra.spark.utils.FilterUtils.parseSSTableTimeRangeFilter; public class ClientConfig @@ -74,6 +76,8 @@ public class ClientConfig public static final String CONSISTENCY_LEVEL_KEY = "consistencyLevel"; public static final String ENABLE_STATS_KEY = "enableStats"; public static final String LAST_MODIFIED_COLUMN_NAME_KEY = "lastModifiedColumnName"; + public static final String CELL_LAST_MODIFIED_COLUMN_NAME_PREFIX = "lastModifiedTimestamp_"; + public static final String CELL_TTL_COLUMN_NAME_PREFIX = "ttl_"; public static final String READ_INDEX_OFFSET_KEY = "readIndexOffset"; public static final String SIZING_KEY = "sizing"; public static final String SIZING_DEFAULT = "default"; @@ -113,7 +117,9 @@ public class ClientConfig protected int maxPartitionSize; protected boolean useIncrementalRepair; protected List requestedFeatures; - protected String lastModifiedTimestampField; + protected String rowLastModifiedTimestampField; + protected Map cellLastModifiedTimestampFields; + protected Map cellTtlFields; protected Boolean enableExpansionShrinkCheck; protected int sidecarPort; protected boolean quoteIdentifiers; @@ -144,7 +150,9 @@ protected ClientConfig(Map options) this.sizing = MapUtils.getOrDefault(options, SIZING_KEY, SIZING_DEFAULT); this.maxPartitionSize = MapUtils.getInt(options, MAX_PARTITION_SIZE_KEY, 1); this.useIncrementalRepair = MapUtils.getBoolean(options, USE_INCREMENTAL_REPAIR, true); - this.lastModifiedTimestampField = MapUtils.getOrDefault(options, LAST_MODIFIED_COLUMN_NAME_KEY, null); + this.rowLastModifiedTimestampField = MapUtils.getOrDefault(options, LAST_MODIFIED_COLUMN_NAME_KEY, null); + this.cellLastModifiedTimestampFields = MapUtils.getKeysWithPrefix(options, CELL_LAST_MODIFIED_COLUMN_NAME_PREFIX, true, null); + this.cellTtlFields = MapUtils.getKeysWithPrefix(options, CELL_TTL_COLUMN_NAME_PREFIX, true, null); this.enableExpansionShrinkCheck = MapUtils.getBoolean(options, ENABLE_EXPANSION_SHRINK_CHECK_KEY, false); this.requestedFeatures = initRequestedFeatures(options); this.sidecarPort = MapUtils.getInt(options, SIDECAR_PORT, DEFAULT_SIDECAR_PORT); @@ -269,9 +277,19 @@ public List requestedFeatures() return requestedFeatures; } - public String lastModifiedTimestampField() + public String rowLastModifiedTimestampField() { - return lastModifiedTimestampField; + return rowLastModifiedTimestampField; + } + + public Map cellLastModifiedTimestampFields() + { + return cellLastModifiedTimestampFields; + } + + public Map cellTtlFields() + { + return cellTtlFields; } public Boolean enableExpansionShrinkCheck() @@ -302,17 +320,33 @@ public static ClientConfig create(Map options) protected List initRequestedFeatures(Map options) { Map optionsCopy = new HashMap<>(options); - String lastModifiedColumnName = MapUtils.getOrDefault(options, LAST_MODIFIED_COLUMN_NAME_KEY, null); - if (lastModifiedColumnName != null) + + String rowLastModifiedColumnName = MapUtils.getOrDefault(options, LAST_MODIFIED_COLUMN_NAME_KEY, null); + if (rowLastModifiedColumnName != null) { optionsCopy.put(SchemaFeatureSet.LAST_MODIFIED_TIMESTAMP.optionName(), "true"); } + Map cellLastModifiedColumns = MapUtils.getKeysWithPrefix(options, CELL_LAST_MODIFIED_COLUMN_NAME_PREFIX, true, null); + if (cellLastModifiedColumns != null) + { + optionsCopy.put(SchemaFeatureSet.CELL_LAST_MODIFIED_TIMESTAMP.optionName(), "true"); + } + Map cellTtlColumns = MapUtils.getKeysWithPrefix(options, CELL_TTL_COLUMN_NAME_PREFIX, true, null); + if (cellTtlColumns != null) + { + optionsCopy.put(SchemaFeatureSet.CELL_TTL.optionName(), "true"); + } + List requestedFeatures = SchemaFeatureSet.initializeFromOptions(optionsCopy); - if (lastModifiedColumnName != null) + + if (rowLastModifiedColumnName != null) { - // Create alias to LAST_MODIFICATION_TIMESTAMP - aliasLastModifiedTimestamp(requestedFeatures, lastModifiedColumnName); + // create an alias, otherwise we use default name + aliasLastModifiedTimestamp(requestedFeatures, rowLastModifiedColumnName); } + addCellLastModifiedTimestamp(requestedFeatures, cellLastModifiedColumns); + addCellTtl(requestedFeatures, cellTtlColumns); + return requestedFeatures; } diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/DataLayer.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/DataLayer.java index bb087dec5..739677b5a 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/DataLayer.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/DataLayer.java @@ -24,6 +24,7 @@ import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Set; import java.util.concurrent.ExecutorService; @@ -94,6 +95,8 @@ public StructType partitionSizeStructType() public StructType structType() { StructType structType = new StructType(); + Set fieldNames = new HashSet<>(); + for (CqlField field : cqlTable().fields()) { // Pass Cassandra field metadata in StructField metadata @@ -102,11 +105,18 @@ public StructType structType() typeConverter().sparkSqlType(field, bigNumberConfig(field)), true, metadata.build()); + fieldNames.add(field.name()); } // Append the requested feature fields for (SchemaFeature feature : requestedFeatures()) { + if (!fieldNames.add(feature.fieldName())) + { + throw new IllegalArgumentException( + String.format("Schema feature field '%s' conflicts with an existing field", + feature.fieldName())); + } feature.generateDataType(cqlTable(), structType); structType = structType.add(feature.field()); } diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/SchemaFeatureCustomizer.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/SchemaFeatureCustomizer.java new file mode 100644 index 000000000..4f4f1a65c --- /dev/null +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/SchemaFeatureCustomizer.java @@ -0,0 +1,211 @@ +/* + * 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.spark.data; + +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; + +import org.apache.cassandra.spark.config.SchemaFeature; +import org.apache.cassandra.spark.config.SchemaFeatureSet; +import org.apache.cassandra.spark.sparksql.CellMetadataDecorator; +import org.apache.cassandra.spark.sparksql.LastModifiedTimestampDecorator; +import org.apache.cassandra.spark.sparksql.RowBuilder; +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.types.DataType; +import org.apache.spark.sql.types.DataTypes; + +class SchemaFeatureCustomizer +{ + private SchemaFeatureCustomizer() + { + throw new IllegalStateException(getClass() + " is static utility class and shall not be instantiated"); + } + + static void aliasLastModifiedTimestamp(List requestedFeatures, String alias) + { + int index = requestedFeatures.indexOf(SchemaFeatureSet.LAST_MODIFIED_TIMESTAMP); + if (index >= 0) + { + SchemaFeature featureAlias = new SchemaFeature() + { + @Override + public String optionName() + { + return SchemaFeatureSet.LAST_MODIFIED_TIMESTAMP.optionName(); + } + + @Override + public String fieldName() + { + return alias; + } + + @Override + public DataType fieldDataType() + { + return SchemaFeatureSet.LAST_MODIFIED_TIMESTAMP.fieldDataType(); + } + + @Override + public RowBuilder decorate(RowBuilder builder) + { + return new LastModifiedTimestampDecorator<>(builder, alias); + } + + @Override + public boolean fieldNullable() + { + return SchemaFeatureSet.LAST_MODIFIED_TIMESTAMP.fieldNullable(); + } + }; + requestedFeatures.set(index, featureAlias); + } + } + + static void addCellLastModifiedTimestamp(List requestedFeatures, Map columns) + { + int index = requestedFeatures.indexOf(SchemaFeatureSet.CELL_LAST_MODIFIED_TIMESTAMP); + if (index >= 0) + { + requestedFeatures.remove(index); + List sortedColumns = columns.keySet().stream().sorted().collect(Collectors.toList()); + for (String column : sortedColumns) + { + SchemaFeature lastModifiedTimestampFeature = new SchemaFeature() + { + @Override + public String optionName() + { + return SchemaFeatureSet.CELL_LAST_MODIFIED_TIMESTAMP.optionName(); + } + + @Override + public String fieldName() + { + return columns.get(column); + } + + @Override + public DataType fieldDataType() + { + return DataTypes.TimestampType; + } + + @Override + public RowBuilder decorate(RowBuilder builder) + { + CqlField source = findTtlAndTimestampAwareCqlField(builder.getCqlTable(), column, optionName()); + return new CellMetadataDecorator<>(builder, source.position(), fieldName(), cell -> cell.timestamp); + } + }; + requestedFeatures.add(lastModifiedTimestampFeature); + } + } + } + + static void addCellTtl(List requestedFeatures, Map columns) + { + int index = requestedFeatures.indexOf(SchemaFeatureSet.CELL_TTL); + if (index >= 0) + { + requestedFeatures.remove(index); + List sortedColumns = columns.keySet().stream().sorted().collect(Collectors.toList()); + for (String column : sortedColumns) + { + SchemaFeature lastModifiedTimestampFeature = new SchemaFeature() + { + @Override + public String optionName() + { + return SchemaFeatureSet.CELL_TTL.optionName(); + } + + @Override + public String fieldName() + { + return columns.get(column); + } + + @Override + public DataType fieldDataType() + { + return DataTypes.IntegerType; + } + + @Override + public RowBuilder decorate(RowBuilder builder) + { + CqlField source = findTtlAndTimestampAwareCqlField(builder.getCqlTable(), column, optionName()); + return new CellMetadataDecorator<>(builder, source.position(), fieldName(), + cell -> cell.ttl == CqlField.NO_TTL ? null : cell.ttl); + } + }; + requestedFeatures.add(lastModifiedTimestampFeature); + } + } + } + + @VisibleForTesting + static CqlField findTtlAndTimestampAwareCqlField(CqlTable table, String sourceColumn, String optionName) + { + // Prefer an exact match first. This is important for quoted identifiers. + CqlField source = table.getField(sourceColumn); + + if (source == null) + { + // Spark options are case-insensitive, so the column suffix may have + // lost its original case. Fall back to case-insensitive resolution. + List matches = table.fields() + .stream() + .filter(field -> field.name().equalsIgnoreCase(sourceColumn)) + .collect(Collectors.toList()); + Preconditions.checkArgument(!matches.isEmpty(), + "Unable to enable schema feature '%s': " + + "column '%s' does not exist in table %s.%s", + optionName, + sourceColumn, + table.keyspace(), + table.table()); + Preconditions.checkArgument(matches.size() == 1, + "Unable to enable schema feature '%s': " + + "column '%s' is ambiguous in table %s.%s. " + + "Matching columns: %s", + optionName, + sourceColumn, + table.keyspace(), + table.table(), + matches.stream() + .map(CqlField::name) + .collect(Collectors.joining(", "))); + source = matches.get(0); + } + + Preconditions.checkArgument(!source.isPrimaryKey(), + "Unable to enable schema feature '%s': " + + "column '%s' is part of primary key", + optionName, + source.name()); + return source; + } +} diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java index 088de9070..fac626f4f 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerTests.java @@ -19,6 +19,8 @@ package org.apache.cassandra.spark.data; +import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -27,7 +29,19 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; +import org.apache.cassandra.bridge.BigNumberConfig; +import org.apache.cassandra.spark.config.SchemaFeature; +import org.apache.cassandra.spark.data.converter.SparkSqlTypeConverter; +import org.apache.spark.sql.types.DataTypes; + import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; class CassandraDataLayerTests { @@ -64,4 +78,52 @@ void testClearSnapshotOptionSupport(Boolean clearSnapshot, String expectedClearS assertThat(clearSnapshotStrategy.hasTTL()).isEqualTo(expectedClearSnapshotStrategy.hasTTL()); assertThat(clearSnapshotStrategy.ttl()).isEqualTo(expectedClearSnapshotStrategy.ttl()); } + + @Test + void testRejectSchemaFeatureFieldConflictingWithTableColumn() + { + CqlTable table = mock(CqlTable.class); + CqlField column = mock(CqlField.class); + SchemaFeature feature = mock(SchemaFeature.class); + SparkSqlTypeConverter typeConverter = mock(SparkSqlTypeConverter.class); + DataLayer dataLayer = mock(DataLayer.class, CALLS_REAL_METHODS); + + when(table.fields()).thenReturn(Collections.singletonList(column)); + + when(column.name()).thenReturn("column1"); + when(column.cqlTypeName()).thenReturn("text"); + + when(typeConverter.sparkSqlType(eq(column), any(BigNumberConfig.class))).thenReturn(DataTypes.StringType); + + when(feature.fieldName()).thenReturn("column1"); + + doReturn(table).when(dataLayer).cqlTable(); + doReturn(typeConverter).when(dataLayer).typeConverter(); + doReturn(Collections.singletonList(feature)).when(dataLayer).requestedFeatures(); + + assertThatThrownBy(dataLayer::structType).isInstanceOf(IllegalArgumentException.class) + .hasMessage("Schema feature field 'column1' conflicts with an existing field"); + } + + @Test + void testRejectDuplicateSchemaFeatureFields() + { + CqlTable table = mock(CqlTable.class); + SchemaFeature ttlFeature = mock(SchemaFeature.class); + SchemaFeature timestampFeature = mock(SchemaFeature.class); + DataLayer dataLayer = mock(DataLayer.class, CALLS_REAL_METHODS); + + when(table.fields()).thenReturn(Collections.emptyList()); + + when(ttlFeature.fieldName()).thenReturn("column1"); + when(ttlFeature.field()).thenReturn(DataTypes.createStructField("column1", DataTypes.IntegerType, true)); + + when(timestampFeature.fieldName()).thenReturn("column1"); + + doReturn(table).when(dataLayer).cqlTable(); + doReturn(Arrays.asList(ttlFeature, timestampFeature)).when(dataLayer).requestedFeatures(); + + assertThatThrownBy(dataLayer::structType).isInstanceOf(IllegalArgumentException.class) + .hasMessage("Schema feature field 'column1' conflicts with an existing field"); + } } diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerValidationTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerValidationTest.java index 68f7b9b1b..d260863e3 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerValidationTest.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/CassandraDataLayerValidationTest.java @@ -267,7 +267,7 @@ private static class TestCassandraDataLayer extends CassandraDataLayer false, // quoteIdentifiers "", // snapshotName null, // datacenter - Sidecar.ClientConfig.create(), // sidecarClientConfig + Sidecar.ClientConfig.create(), // sidecarClientConfig null, // sslConfig mock(CqlTable.class), // cqlTable mock(TokenPartitioner.class), // tokenPartitioner @@ -280,7 +280,9 @@ private static class TestCassandraDataLayer extends CassandraDataLayer false, // enableStats false, // readIndexOffset false, // useIncrementalRepair - null, // lastModifiedTimestampField + null, // rowLastModifiedTimestampField + null, // cellLastModifiedTimestampFields + null, // cellTtlFields Collections.emptyList(), // requestedFeatures Collections.emptyMap(), // rfMap mock(TimeProvider.class), // timeProvider diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/SchemaFeatureCustomizerTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/SchemaFeatureCustomizerTest.java new file mode 100644 index 000000000..1583ee83b --- /dev/null +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/SchemaFeatureCustomizerTest.java @@ -0,0 +1,110 @@ +/* + * 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.spark.data; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.junit.jupiter.api.Test; + +import static org.apache.cassandra.spark.data.SchemaFeatureCustomizer.findTtlAndTimestampAwareCqlField; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class SchemaFeatureCustomizerTest +{ + @Test + void testFindCqlFieldExactMatch() + { + CqlTable table = mock(CqlTable.class); + CqlField column1 = mock(CqlField.class); + + when(table.getField("column1")).thenReturn(column1); + when(column1.isPrimaryKey()).thenReturn(false); + + CqlField result = findTtlAndTimestampAwareCqlField(table, "column1", "cell_ttl"); + assertThat(result).isSameAs(column1); + } + + @Test + void testFindCqlFieldCaseInsensitiveMatch() + { + CqlTable table = mock(CqlTable.class); + CqlField column = new CqlField(false, false, false, "Column1", mock(CqlField.CqlType.class), 0); + + when(table.getField("column1")).thenReturn(null); + when(table.fields()).thenReturn(Collections.singletonList(column)); + + CqlField result = findTtlAndTimestampAwareCqlField(table, "column1", "cell_ttl"); + assertThat(result).isSameAs(column); + } + + @Test + void testFindCqlFieldRejectsMissingColumn() + { + CqlTable table = mock(CqlTable.class); + CqlField column1 = new CqlField(false, false, false, "a", mock(CqlField.CqlType.class), 0); + CqlField column2 = new CqlField(false, false, false, "b", mock(CqlField.CqlType.class), 1); + + when(table.getField("missing")).thenReturn(null); + when(table.fields()).thenReturn(List.of(column1, column2)); + when(table.keyspace()).thenReturn("ks"); + when(table.table()).thenReturn("tbl"); + + assertThatThrownBy(() -> findTtlAndTimestampAwareCqlField(table, "missing", "cell_ttl")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Unable to enable schema feature 'cell_ttl': column 'missing' does not exist in table ks.tbl"); + } + + @Test + void testFindCqlFieldRejectsAmbiguousCaseInsensitiveMatch() + { + CqlTable table = mock(CqlTable.class); + CqlField upperCaseColumn = new CqlField(false, false, false, "Column", mock(CqlField.CqlType.class), 0); + CqlField lowerCaseColumn = new CqlField(false, false, false, "column", mock(CqlField.CqlType.class), 0); + + when(table.getField("COLUMN")).thenReturn(null); + when(table.fields()).thenReturn(Arrays.asList(upperCaseColumn, lowerCaseColumn)); + when(table.keyspace()).thenReturn("ks"); + when(table.table()).thenReturn("tbl"); + + assertThatThrownBy(() -> findTtlAndTimestampAwareCqlField(table, "COLUMN", "cell_ttl")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Unable to enable schema feature 'cell_ttl': column 'COLUMN' is ambiguous in table ks.tbl. " + + "Matching columns: Column, column"); + } + + @Test + void testFindCqlFieldRejectsPrimaryKey() + { + CqlTable table = mock(CqlTable.class); + CqlField idColumn = new CqlField(true, false, false, "id", mock(CqlField.CqlType.class), 0); + + when(table.getField("id")).thenReturn(idColumn); + + assertThatThrownBy(() -> findTtlAndTimestampAwareCqlField(table, "id", "cell_ttl")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Unable to enable schema feature 'cell_ttl'") + .hasMessageContaining("column 'id' is part of primary key"); + } +} diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/TimestampIntegrationTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/TimestampTtlIntegrationTest.java similarity index 53% rename from cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/TimestampIntegrationTest.java rename to cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/TimestampTtlIntegrationTest.java index 98b96d62a..b2d132d12 100644 --- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/TimestampIntegrationTest.java +++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/TimestampTtlIntegrationTest.java @@ -20,6 +20,7 @@ package org.apache.cassandra.analytics; import java.time.Instant; +import java.util.ArrayList; import java.util.Arrays; import java.util.Comparator; import java.util.List; @@ -34,6 +35,7 @@ import org.apache.cassandra.sidecar.testing.QualifiedName; import org.apache.cassandra.spark.bulkwriter.TimestampOption; import org.apache.cassandra.spark.bulkwriter.WriterOptions; +import org.apache.cassandra.spark.data.CqlField; import org.apache.cassandra.testing.ClusterBuilderConfiguration; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; @@ -45,14 +47,30 @@ import static org.assertj.core.api.Assertions.assertThat; /** - * Integration test for the Cassandra timestamps + * Integration test for the Cassandra timestamps and TTLs */ -class TimestampIntegrationTest extends SharedClusterSparkIntegrationTestBase +class TimestampTtlIntegrationTest extends SharedClusterSparkIntegrationTestBase { static final List DATASET = Arrays.asList("a", "b", "c", "d", "e", "f", "g"); + + // table contains rows with custom TIMESTAMP and TTL static final QualifiedName SOURCE_TABLE = uniqueTestTableFullName(TEST_KEYSPACE, "source_tbl"); + + // table contains rows with custom TIMESTAMP only + static final QualifiedName NO_TTL_TABLE = uniqueTestTableFullName(TEST_KEYSPACE, "no_ttl_tbl"); + + // table contains rows whose cells contain different TTL value + static final QualifiedName VARIABLE_TTL_TABLE = uniqueTestTableFullName(TEST_KEYSPACE, "variable_ttl_tbl"); + static final QualifiedName TARGET_TABLE = uniqueTestTableFullName(TEST_KEYSPACE, "target_tbl"); - static final List TABLE_NAMES = Arrays.asList(SOURCE_TABLE, TARGET_TABLE); + + static final List TABLE_NAMES = Arrays.asList(SOURCE_TABLE, + TARGET_TABLE, + NO_TTL_TABLE, + VARIABLE_TTL_TABLE); + + static final long desiredTimestamp = 1432815430948567L; + static final int desiredTtl = 600; /** * Reads from source table with timestamps, and then persist the read data to the target @@ -73,15 +91,93 @@ void testReadingAndWritingTimestamp() validateWrites(TARGET_TABLE, rowList); } + @Test + void testReadingCellTimestampAndTtl() throws Exception + { + Thread.sleep(2000); // elapse two seconds so that TTL differs + + Dataset data = bulkReaderDataFrame(SOURCE_TABLE).option("lastModifiedTimestamp_course", "courseTimestamp") + .option("ttl_course", "courseTtl") + .option("lastModifiedTimestamp_marks", "marksTimestamp") + .option("ttl_marks", "marksTtl") + .load() + .select("id", "courseTimestamp", "courseTtl", "marksTimestamp", "marksTtl"); + + List rows = data.collectAsList(); + + assertThat(rows).hasSize(DATASET.size()); + + rows.forEach(row -> { + Instant timestamp = row.getTimestamp(1).toInstant(); + assertThat(timestamp.getEpochSecond()).isEqualTo(1432815430L); + assertThat(timestamp.getNano()).isEqualTo(948567000L); + + int ttl = row.getInt(2); + assertThat(ttl).isBetween(1, 599); + + assertThat(row.getTimestamp(3)).isNotNull(); + assertThat(row.getInt(4)).isNotNull(); + }); + } + + @Test + void testReadingCellWithoutTtl() throws Exception + { + populateTable(NO_TTL_TABLE, DATASET, desiredTimestamp, CqlField.NO_TTL); + Dataset data = bulkReaderDataFrame(NO_TTL_TABLE).option("ttl_course", "courseTtl") + .option("ttl_marks", "marksTtl") + .load() + .select("id", "courseTtl", "marksTtl"); + + List rows = data.collectAsList(); + + assertThat(rows).hasSize(DATASET.size()); + + rows.forEach(row -> { + assertThat(row.isNullAt(1)).isTrue(); + assertThat(row.isNullAt(2)).isTrue(); + }); + } + + @Test + void testReadingRowWithVariableTtlAndTimestamp() throws Exception + { + ICoordinator coordinator = cluster.getFirstRunningInstance().coordinator(); + String query = String.format("INSERT INTO %s (id, course, marks) VALUES (%d,'%s',%d) USING TTL %d", + VARIABLE_TTL_TABLE, 1, "course_a", 2, desiredTtl); + coordinator.execute(query, ConsistencyLevel.ALL); + + // update TTL of "marks" column for TTL to differ + query = String.format("UPDATE %s USING TTL %d SET marks = %d WHERE id = %d", + VARIABLE_TTL_TABLE, desiredTtl / 2, 3, 1); + Thread.sleep(2000); + coordinator.execute(query, ConsistencyLevel.ALL); + + Dataset data = bulkReaderDataFrame(VARIABLE_TTL_TABLE).option("lastModifiedTimestamp_course", "courseTimestamp") + .option("ttl_course", "courseTtl") + .option("lastModifiedTimestamp_marks", "marksTimestamp") + .option("ttl_marks", "marksTtl") + .load() + .select("id", "courseTimestamp", "courseTtl", "marksTimestamp", "marksTtl"); + + List rows = data.collectAsList(); + + assertThat(rows).hasSize(1); + Row row = rows.get(0); + // write timestamp of "course" should be earlier than "marks" + assertThat(row.getTimestamp(1).toInstant()).isBefore(row.getTimestamp(3).toInstant()); + // TTL of "marks" column has been decreased with UPDATE statement + assertThat(row.getInt(2)).isGreaterThan(row.getInt(4)); + } + @Override protected void initializeSchemaForTest() { - long desiredTimestamp = 1432815430948567L; TABLE_NAMES.forEach(name -> { createTestKeyspace(name, DC1_RF1); createTestTable(name, CREATE_TEST_TABLE_STATEMENT); }); - populateTable(SOURCE_TABLE, DATASET, desiredTimestamp); + populateTable(SOURCE_TABLE, DATASET, desiredTimestamp, desiredTtl); } @Override @@ -127,14 +223,20 @@ void validateWrites(QualifiedName tableName, List sourceData) .isEmpty(); } - void populateTable(QualifiedName tableName, List values, long desiredTimestamp) + void populateTable(QualifiedName tableName, List values, long desiredTimestamp, int desiredTtl) { ICoordinator coordinator = cluster.getFirstRunningInstance().coordinator(); for (int i = 0; i < values.size(); i++) { String value = values.get(i); - String query = String.format("INSERT INTO %s (id, course, marks) VALUES (%d,'%s',%d) USING TIMESTAMP %d", - tableName, i, "course_" + value, 80 + i, desiredTimestamp); + String query = "INSERT INTO %s (id, course, marks) VALUES (%d,'%s',%d) USING TIMESTAMP %d"; + List variables = new ArrayList<>(Arrays.asList(tableName, i, "course_" + value, 80 + i, desiredTimestamp)); + if (desiredTtl != CqlField.NO_TTL) + { + query += " AND TTL %d"; + variables.add(desiredTtl); + } + query = String.format(query, variables.toArray()); coordinator.execute(query, ConsistencyLevel.ALL); } } diff --git a/cassandra-analytics-spark-converter/src/main/java/org/apache/cassandra/spark/config/SchemaFeatureSet.java b/cassandra-analytics-spark-converter/src/main/java/org/apache/cassandra/spark/config/SchemaFeatureSet.java index 671fbdccb..2bfe35bd2 100644 --- a/cassandra-analytics-spark-converter/src/main/java/org/apache/cassandra/spark/config/SchemaFeatureSet.java +++ b/cassandra-analytics-spark-converter/src/main/java/org/apache/cassandra/spark/config/SchemaFeatureSet.java @@ -48,6 +48,38 @@ public RowBuilder decorate(RowBuilder builder) { return new LastModifiedTimestampDecorator<>(builder, fieldName()); } + }, + + // Special column that passes over last modified timestamp for a cell + CELL_LAST_MODIFIED_TIMESTAMP + { + @Override + public DataType fieldDataType() + { + throw new UnsupportedOperationException(); + } + + @Override + public RowBuilder decorate(RowBuilder builder) + { + throw new UnsupportedOperationException(); + } + }, + + // Special column that passes over remaining TTL for a cell + CELL_TTL + { + @Override + public DataType fieldDataType() + { + throw new UnsupportedOperationException(); + } + + @Override + public RowBuilder decorate(RowBuilder builder) + { + throw new UnsupportedOperationException(); + } }; /** diff --git a/cassandra-analytics-spark-converter/src/main/java/org/apache/cassandra/spark/sparksql/CellMetadataDecorator.java b/cassandra-analytics-spark-converter/src/main/java/org/apache/cassandra/spark/sparksql/CellMetadataDecorator.java new file mode 100644 index 000000000..c85e53dcf --- /dev/null +++ b/cassandra-analytics-spark-converter/src/main/java/org/apache/cassandra/spark/sparksql/CellMetadataDecorator.java @@ -0,0 +1,82 @@ +/* + * 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.spark.sparksql; + +import java.util.function.Function; + +import org.apache.spark.sql.catalyst.InternalRow; + +/** + * Wrapper allowing to append any cell attribute to Spark row. + * @param type of row returned by this builder + */ +public class CellMetadataDecorator extends RowBuilderDecorator +{ + private final int sourceColumnPosition; + private final int metadataColumnPosition; + private final Function metadataGetter; + private Object metadata; + + public CellMetadataDecorator(RowBuilder delegate, + int sourceColumnPosition, + String fieldName, + Function metadataGetter) + { + super(delegate); + this.sourceColumnPosition = sourceColumnPosition; + this.metadataGetter = metadataGetter; + + int width = internalExpandRow(); + int fieldIndex = fieldIndex(fieldName); + this.metadataColumnPosition = fieldIndex >= 0 ? fieldIndex : width; + } + + @Override + public void reset() + { + super.reset(); + metadata = null; + } + + @Override + public void onCell(Cell cell) + { + super.onCell(cell); + if (cell.isPkCkOnly || cell.position != sourceColumnPosition) + { + return; + } + // apply metadata only to non-primary key columns only + metadata = metadataGetter.apply(cell); + } + + @Override + protected int extraColumns() + { + return 1; + } + + @Override + public T build() + { + array()[metadataColumnPosition] = metadata; + return super.build(); + } +} diff --git a/cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/spark/reader/AbstractStreamScanner.java b/cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/spark/reader/AbstractStreamScanner.java index 202fe89f5..745cc8240 100644 --- a/cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/spark/reader/AbstractStreamScanner.java +++ b/cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/spark/reader/AbstractStreamScanner.java @@ -25,6 +25,7 @@ import java.util.Iterator; import com.google.common.base.Preconditions; +import com.google.common.primitives.Ints; import org.apache.cassandra.db.Clustering; import org.apache.cassandra.db.ClusteringPrefix; @@ -40,6 +41,7 @@ import org.apache.cassandra.dht.RandomPartitioner; import org.apache.cassandra.schema.ColumnMetadata; import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.spark.data.CqlField; import org.apache.cassandra.spark.data.partitioner.Partitioner; import org.apache.cassandra.spark.reader.common.SSTableStreamException; import org.apache.cassandra.spark.utils.TimeProvider; @@ -395,6 +397,15 @@ public void consume() rowData.setValueCopy(cell.buffer()); } rowData.setTimestamp(cell.timestamp()); + if (cell.isExpiring()) + { + long remaining = cell.localDeletionTime() - timeProvider.referenceEpochInSeconds(); + rowData.setTtl(Ints.checkedCast(remaining)); + } + else + { + rowData.setTtl(CqlField.NO_TTL); + } // Null out clustering so hasData will return false clustering = null; } @@ -438,6 +449,8 @@ public void consume() { AbstractComplexTypeBuffer buffer = AbstractComplexTypeBuffer.newBuffer(column.type, cellCount); long maxTimestamp = Long.MIN_VALUE; + // from non-frozen complex types, we return the lowest TTL that will modify the type's state + long minTtl = CqlField.NO_TTL; while (cells.hasNext()) { Cell cell = cells.next(); @@ -453,16 +466,23 @@ public void consume() } // In the case the cell is deleted, the deletion time is also the cell's timestamp maxTimestamp = Math.max(maxTimestamp, cell.timestamp()); + if (cell.isExpiring()) + { + long remaining = cell.localDeletionTime() - timeProvider.referenceEpochInSeconds(); + minTtl = minTtl == CqlField.NO_TTL || remaining < 0 ? remaining : Math.min(remaining, minTtl); + } } rowData.setValueCopy(buffer.build()); rowData.setTimestamp(maxTimestamp); + rowData.setTtl(Ints.checkedCast(minTtl)); } else { // The entire collection/UDT is deleted handleCellTombstone(rowData.getToken()); rowData.setTimestamp(deletionTime.markedForDeleteAt()); + rowData.setTtl(CqlField.NO_TTL); } // Null out clustering to indicate no data diff --git a/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/spark/reader/AbstractStreamScanner.java b/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/spark/reader/AbstractStreamScanner.java index 2e7d30d03..2a4901934 100644 --- a/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/spark/reader/AbstractStreamScanner.java +++ b/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/spark/reader/AbstractStreamScanner.java @@ -41,6 +41,7 @@ import org.apache.cassandra.dht.RandomPartitioner; import org.apache.cassandra.schema.ColumnMetadata; import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.spark.data.CqlField; import org.apache.cassandra.spark.data.partitioner.Partitioner; import org.apache.cassandra.spark.reader.common.SSTableStreamException; import org.apache.cassandra.spark.utils.TimeProvider; @@ -396,6 +397,15 @@ public void consume() rowData.setValueCopy(cell.buffer()); } rowData.setTimestamp(cell.timestamp()); + if (cell.isExpiring()) + { + long remaining = cell.localDeletionTime() - timeProvider.referenceEpochInSeconds(); + rowData.setTtl(Ints.checkedCast(remaining)); + } + else + { + rowData.setTtl(CqlField.NO_TTL); + } // Null out clustering so hasData will return false clustering = null; } @@ -439,6 +449,8 @@ public void consume() { AbstractComplexTypeBuffer buffer = AbstractComplexTypeBuffer.newBuffer(column.type, cellCount); long maxTimestamp = Long.MIN_VALUE; + // from non-frozen complex types, we return the lowest TTL that will modify the type's state + long minTtl = CqlField.NO_TTL; // C* 4.0 Cell.isLive requires int for nowInSec; checked cast will throw after Y2038 int referenceEpochInSecondsAsInt = Ints.checkedCast(timeProvider.referenceEpochInSeconds()); while (cells.hasNext()) @@ -456,16 +468,23 @@ public void consume() } // In the case the cell is deleted, the deletion time is also the cell's timestamp maxTimestamp = Math.max(maxTimestamp, cell.timestamp()); + if (cell.isExpiring()) + { + long remaining = cell.localDeletionTime() - timeProvider.referenceEpochInSeconds(); + minTtl = minTtl == CqlField.NO_TTL || remaining < 0 ? remaining : Math.min(remaining, minTtl); + } } rowData.setValueCopy(buffer.build()); rowData.setTimestamp(maxTimestamp); + rowData.setTtl(Ints.checkedCast(minTtl)); } else { // The entire collection/UDT is deleted handleCellTombstone(rowData.getToken()); rowData.setTimestamp(deletionTime.markedForDeleteAt()); + rowData.setTtl(CqlField.NO_TTL); } // Null out clustering to indicate no data diff --git a/docs/src/user.adoc b/docs/src/user.adoc index 80ae10ab3..490e3bba7 100644 --- a/docs/src/user.adoc +++ b/docs/src/user.adoc @@ -167,7 +167,25 @@ with the following structure: |_lastModifiedColumnName_ |no | -|Name of the field to be appended to Spark RDD that represents last modification timestamp of each row +|Name of the field to be appended to Spark RDD that represents last modification timestamp of each row. +The timestamp is the maximum write timestamp across the cells in the row. + +|_++lastModifiedTimestamp_++_ +|no +| +a|Adds a field containing the last modification timestamp of the specified Cassandra column. + +The Cassandra column name is specified as a suffix of the property name, and the property value specifies the +name of the field to append to the Spark schema. + +For example, `lastModifiedTimestamp_my_column = myColumnTsmp` reads the write timestamp of the `my_column` column +and exposes it as the `myColumnTsmp` timestamp field. + +|_++ttl_++_ +|no +| +|Adds a field containing the remaining time-to-live (in seconds) of the specified Cassandra column. +Usage of the property follows the same semantics as `lastModifiedTimestamp_` option. |===