Add optional TTL for migrated Cassandra timeseries data

README.md

@@ -336,6 +336,26 @@ sstableloader --verbose --nodes CASSANDRA_NODES --username cassandra --password
 
 Verify that historical data available in ThingsBoard.
 
+# Setting TTL for migrated data
+
+By default migrated data never expires. To make migrated timeseries data expire automatically in Cassandra,
+pass the optional `-ttl` argument (in days) when running the tool:
+
+```
+java -jar ./target/database-migrator-1.0-SNAPSHOT-jar-with-dependencies.jar \
+        ... \
+        -ttl 90
+```
+
+Notes:
+* TTL applies only to `ts_kv_cf` and `ts_kv_partitions_cf` (historical data points and their partition bookkeeping).
+* `ts_kv_latest_cf` (last known value per key) is never affected by `-ttl` — this matches ThingsBoard's own TTL
+  behavior, where only historical points expire, not the latest value.
+* TTL is counted from the moment the SSTables are generated by this tool, not from the original timestamp of
+  the data being migrated.
+* `-ttl` must be a positive number of days and cannot exceed 7300 days (20 years), which is Cassandra's
+  hard-coded maximum TTL.
+
 # Troubleshooting
 
 ## Continue migration in case of failure on particular migration line

src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java

@@ -25,10 +25,14 @@
 import org.apache.commons.cli.ParseException;
 
 import java.io.File;
+import java.util.concurrent.TimeUnit;
 
 @Slf4j
 public class MigratorTool {
 
+    private static final long MAX_TTL_DAYS = 7300; // Cassandra's hard-coded max TTL is 20 years
+    private static final long MAX_TTL_SECONDS = TimeUnit.DAYS.toSeconds(MAX_TTL_DAYS);
+
     public static void main(String[] args) {
         CommandLine cmd = parseArgs(args);
 
@@ -60,6 +64,19 @@ public static void main(String[] args) {
                 linesToSkip = Integer.parseInt(cmd.getOptionValue("linesToSkip"));
             }
 
+            Long ttlSeconds = null;
+            if (cmd.getOptionValue("ttl") != null) {
+                long ttlDays = Long.parseLong(cmd.getOptionValue("ttl"));
+                if (ttlDays <= 0) {
+                    throw new RuntimeException("Failed to parse ttl property: ttl must be a positive number of days!");
+                }
+                ttlSeconds = TimeUnit.DAYS.toSeconds(ttlDays);
+                if (ttlSeconds > MAX_TTL_SECONDS) {
+                    throw new RuntimeException("Failed to parse ttl property: ttl of " + ttlDays +
+                            " days exceeds Cassandra's maximum allowed TTL of " + MAX_TTL_DAYS + " days (20 years)!");
+                }
+            }
+
             new PgCaMigrator(
                     allTelemetrySource,
                     tsSaveDir,
@@ -68,7 +85,8 @@ public static void main(String[] args) {
                     allEntityIdsAndTypes,
                     dictionaryParser,
                     castEnable,
-                    partitioning).migrate(linesToSkip);
+                    partitioning,
+                    ttlSeconds).migrate(linesToSkip);
 
         } catch (Throwable th) {
             log.error("Failed to migrate", th);
@@ -117,6 +135,13 @@ private static CommandLine parseArgs(String[] args) {
         linesToSkipOpt.setRequired(false);
         options.addOption(linesToSkipOpt);
 
+        Option ttlOpt = new Option("ttl", "ttl", true,
+                "TTL in days for migrated timeseries data (ts_kv_cf and ts_kv_partitions_cf). " +
+                        "Must be a positive number not exceeding 7300 days (Cassandra's 20 year TTL limit). " +
+                        "If not set, migrated data will never expire. Does not affect ts_kv_latest_cf.");
+        ttlOpt.setRequired(false);
+        options.addOption(ttlOpt);
+
         HelpFormatter formatter = new HelpFormatter();
         CommandLineParser parser = new BasicParser();
 

src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java

@@ -39,13 +39,14 @@ public PgCaMigrator(File sourceFile,
                         RelatedEntitiesParser allEntityIdsAndTypes,
                         DictionaryParser dictionaryParser,
                         boolean castStringsIfPossible,
-                        String partitioning) {
+                        String partitioning,
+                        Long ttlSeconds) {
         this.sourceFile = sourceFile;
         if (outTsLatestDir != null) {
             this.tbLatestWriter = new TbLatestWriter(dictionaryParser, allEntityIdsAndTypes, outTsLatestDir, castStringsIfPossible, partitioning);
         }
         if (ourTsDir != null) {
-            this.tbTsWriter = new TbTsWriter(dictionaryParser, allEntityIdsAndTypes, ourTsDir, outTsPartitionDir, castStringsIfPossible, partitioning);
+            this.tbTsWriter = new TbTsWriter(dictionaryParser, allEntityIdsAndTypes, ourTsDir, outTsPartitionDir, castStringsIfPossible, partitioning, ttlSeconds);
         }
     }
 

src/main/java/org/thingsboard/client/tools/migrator/WriterBuilder.java

@@ -57,15 +57,17 @@ public class WriterBuilder {
             ") WITH CLUSTERING ORDER BY ( partition ASC )\n" +
             "  AND compaction = { 'class' :  'LeveledCompactionStrategy'  };";
 
-    public static CQLSSTableWriter getTsWriter(File dir) {
+    public static CQLSSTableWriter getTsWriter(File dir, Long ttlSeconds) {
         return CQLSSTableWriter.builder()
                 .inDirectory(dir.getAbsolutePath())
                 .forTable(tsSchema)
                 .using("INSERT INTO thingsboard.ts_kv_cf (entity_type, entity_id, key, partition, ts, bool_v, str_v, long_v, dbl_v, json_v) " +
-                        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
+                        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" + ttlClause(ttlSeconds))
                 .build();
     }
 
+    // Latest values intentionally never expire, regardless of ttlSeconds - they represent
+    // the current state of a key, not historical points, matching ThingsBoard's own TTL semantics.
     public static CQLSSTableWriter getLatestWriter(File dir) {
         return CQLSSTableWriter.builder()
                 .inDirectory(dir.getAbsolutePath())
@@ -75,12 +77,16 @@ public static CQLSSTableWriter getLatestWriter(File dir) {
                 .build();
     }
 
-    public static CQLSSTableWriter getPartitionWriter(File dir) {
+    public static CQLSSTableWriter getPartitionWriter(File dir, Long ttlSeconds) {
         return CQLSSTableWriter.builder()
                 .inDirectory(dir.getAbsolutePath())
                 .forTable(partitionSchema)
                 .using("INSERT INTO thingsboard.ts_kv_partitions_cf (entity_type, entity_id, key, partition) " +
-                        "VALUES (?, ?, ?, ?)")
+                        "VALUES (?, ?, ?, ?)" + ttlClause(ttlSeconds))
                 .build();
     }
+
+    private static String ttlClause(Long ttlSeconds) {
+        return ttlSeconds != null ? " USING TTL " + ttlSeconds : "";
+    }
 }

src/main/java/org/thingsboard/client/tools/migrator/writer/AbstractTbWriter.java

@@ -67,10 +67,14 @@ public abstract class AbstractTbWriter implements TbWriter {
 
     private final boolean castStringIfPossible;
 
+    // null means no TTL - migrated rows never expire, matching pre-existing behavior
+    protected final Long ttlSeconds;
+
     public AbstractTbWriter(DictionaryParser keyParser, RelatedEntitiesParser entityIdsAndTypes, File outDir,
-                            boolean castStringIfPossible, String partitioning) {
+                            boolean castStringIfPossible, String partitioning, Long ttlSeconds) {
         this.keyParser = keyParser;
         this.entityIdsAndTypes = entityIdsAndTypes;
+        this.ttlSeconds = ttlSeconds;
         this.currentWriter = getWriter(outDir);
         this.outDir = outDir;
         this.castStringIfPossible = castStringIfPossible;

src/main/java/org/thingsboard/client/tools/migrator/writer/TbLatestWriter.java

@@ -29,7 +29,8 @@ public class TbLatestWriter extends AbstractTbWriter {
 
     public TbLatestWriter(DictionaryParser keyParser, RelatedEntitiesParser entityIdsAndTypes, File outDir,
                           boolean castStringsIfPossible, String partitioning) {
-        super(keyParser, entityIdsAndTypes, outDir, castStringsIfPossible, partitioning);
+        // latest values never expire regardless of the migration's ttl setting
+        super(keyParser, entityIdsAndTypes, outDir, castStringsIfPossible, partitioning, null);
     }
 
     @Override

src/main/java/org/thingsboard/client/tools/migrator/writer/TbTsWriter.java

@@ -34,8 +34,8 @@ public class TbTsWriter extends AbstractTbWriter {
     private final File outTsPartitionDir;
 
     public TbTsWriter(DictionaryParser keyParser, RelatedEntitiesParser entityIdsAndTypes, File outDir,
-                      File outTsPartitionDir, boolean castStringsIfPossible, String partitioning) {
-        super(keyParser, entityIdsAndTypes, outDir, castStringsIfPossible, partitioning);
+                      File outTsPartitionDir, boolean castStringsIfPossible, String partitioning, Long ttlSeconds) {
+        super(keyParser, entityIdsAndTypes, outDir, castStringsIfPossible, partitioning, ttlSeconds);
         this.outTsPartitionDir = outTsPartitionDir;
     }
 
@@ -61,19 +61,19 @@ public List<Object> toValues(List<String> raw) {
     @Override
     public void reOpenWriter() throws IOException {
         currentWriter.close();
-        currentWriter = WriterBuilder.getTsWriter(outDir);
+        currentWriter = WriterBuilder.getTsWriter(outDir, ttlSeconds);
     }
 
     @Override
     public CQLSSTableWriter getWriter(File outDir) {
-        return WriterBuilder.getTsWriter(outDir);
+        return WriterBuilder.getTsWriter(outDir, ttlSeconds);
     }
 
     @Override
     public void writePartitions() throws IOException {
         CQLSSTableWriter currentPartitionsWriter = null;
         try {
-            currentPartitionsWriter = WriterBuilder.getPartitionWriter(outTsPartitionDir);
+            currentPartitionsWriter = WriterBuilder.getPartitionWriter(outTsPartitionDir, ttlSeconds);
             log.info("Partitions collected " + partitions.size());
             long startTs = System.currentTimeMillis();
             for (String partition : partitions) {