Timeseries model

ATTRIBUTIONS.md

@@ -145,6 +145,12 @@ The following table lists runtime dependencies bundled with ArcadeDB distributio
 
 **License Note:** UPL = Universal Permissive License 1.0
 
+### Compression
+
+| Group ID | Artifact ID | Version | License | Homepage |
+|----------|-------------|---------|---------|----------|
+| org.xerial.snappy | snappy-java | 1.1.10.7 | Apache 2.0 | https://github.com/xerial/snappy-java |
+
 ### Server and Networking
 
 | Group ID | Artifact ID | Version | License | Homepage |

engine/src/main/antlr4/com/arcadedb/query/sql/grammar/SQLLexer.g4

@@ -231,6 +231,20 @@ MINUTE: M I N U T E;
 HOUR: H O U R;
 MANUAL: M A N U A L;
 INCREMENTAL: I N C R E M E N T A L;
+TIMESERIES: T I M E S E R I E S;
+TAGS: T A G S;
+FIELDS: F I E L D S;
+RETENTION: R E T E N T I O N;
+COMPACTION_INTERVAL: C O M P A C T I O N UNDERSCORE I N T E R V A L;
+SHARDS: S H A R D S;
+DAYS: D A Y S;
+HOURS: H O U R S;
+MINUTES: M I N U T E S;
+DOWNSAMPLING: D O W N S A M P L I N G;
+POLICY: P O L I C Y;
+GRANULARITY: G R A N U L A R I T Y;
+CONTINUOUS: C O N T I N U O U S;
+AGGREGATE: A G G R E G A T E;
 
 // ============================================================================
 // COMPARISON OPERATORS

engine/src/main/antlr4/com/arcadedb/query/sql/grammar/SQLParser.g4

@@ -95,13 +95,16 @@ statement
     | CREATE EDGE createEdgeBody                     # createEdgeStmt
     | CREATE TRIGGER createTriggerBody               # createTriggerStmt
     | CREATE MATERIALIZED VIEW createMaterializedViewBody   # createMaterializedViewStmt
+    | CREATE TIMESERIES TYPE createTimeSeriesTypeBody       # createTimeSeriesTypeStmt
+    | CREATE CONTINUOUS AGGREGATE createContinuousAggregateBody  # createContinuousAggregateStmt
 
     // DDL Statements - ALTER variants
     | ALTER TYPE alterTypeBody                       # alterTypeStmt
     | ALTER PROPERTY alterPropertyBody               # alterPropertyStmt
     | ALTER BUCKET alterBucketBody                   # alterBucketStmt
     | ALTER DATABASE alterDatabaseBody               # alterDatabaseStmt
     | ALTER MATERIALIZED VIEW alterMaterializedViewBody     # alterMaterializedViewStmt
+    | ALTER TIMESERIES TYPE alterTimeSeriesTypeBody        # alterTimeSeriesTypeStmt
 
     // DDL Statements - DROP variants
     | DROP TYPE dropTypeBody                         # dropTypeStmt
@@ -110,6 +113,7 @@ statement
     | DROP BUCKET dropBucketBody                     # dropBucketStmt
     | DROP TRIGGER dropTriggerBody                   # dropTriggerStmt
     | DROP MATERIALIZED VIEW dropMaterializedViewBody       # dropMaterializedViewStmt
+    | DROP CONTINUOUS AGGREGATE dropContinuousAggregateBody # dropContinuousAggregateStmt
 
     // DDL Statements - TRUNCATE variants
     | TRUNCATE TYPE truncateTypeBody                 # truncateTypeStmt
@@ -119,6 +123,9 @@ statement
     // Materialized View Refresh
     | REFRESH MATERIALIZED VIEW refreshMaterializedViewBody # refreshMaterializedViewStmt
 
+    // Continuous Aggregate Refresh
+    | REFRESH CONTINUOUS AGGREGATE refreshContinuousAggregateBody # refreshContinuousAggregateStmt
+
     // Index Management
     | rebuildIndexStatement                          # rebuildIndexStmt
 
@@ -425,6 +432,51 @@ createTypeBody
       (PAGESIZE INTEGER_LITERAL)?
     ;
 
+/**
+ * CREATE TIMESERIES TYPE body
+ * Example: CREATE TIMESERIES TYPE SensorData TIMESTAMP ts TAGS (sensor_id STRING) FIELDS (temperature DOUBLE, humidity DOUBLE) SHARDS 4 RETENTION 90 DAYS COMPACTION_INTERVAL 1 HOURS
+ */
+createTimeSeriesTypeBody
+    : identifier
+      (IF NOT EXISTS)?
+      (TIMESTAMP identifier)?
+      (TAGS LPAREN tsTagColumnDef (COMMA tsTagColumnDef)* RPAREN)?
+      (FIELDS LPAREN tsFieldColumnDef (COMMA tsFieldColumnDef)* RPAREN)?
+      (SHARDS INTEGER_LITERAL)?
+      (RETENTION INTEGER_LITERAL (DAYS | HOURS | MINUTES)?)?
+      (COMPACTION_INTERVAL INTEGER_LITERAL (DAYS | HOURS | MINUTES)?)?
+    ;
+
+tsTagColumnDef
+    : identifier identifier
+    ;
+
+tsFieldColumnDef
+    : identifier identifier
+    ;
+
+/**
+ * ALTER TIMESERIES TYPE body - add or drop downsampling policy
+ * Example: ALTER TIMESERIES TYPE SensorData ADD DOWNSAMPLING POLICY AFTER 7 DAYS GRANULARITY 1 HOURS AFTER 30 DAYS GRANULARITY 1 DAYS
+ * Example: ALTER TIMESERIES TYPE SensorData DROP DOWNSAMPLING POLICY
+ */
+alterTimeSeriesTypeBody
+    : identifier ADD DOWNSAMPLING POLICY downsamplingTierClause+
+    | identifier DROP DOWNSAMPLING POLICY
+    ;
+
+downsamplingTierClause
+    : AFTER INTEGER_LITERAL tsTimeUnit GRANULARITY INTEGER_LITERAL tsTimeUnit
+    ;
+
+tsTimeUnit
+    : DAYS
+    | HOURS
+    | MINUTES
+    | HOUR
+    | MINUTE
+    ;
+
 /**
  * CREATE EDGE TYPE body (supports UNIDIRECTIONAL)
  */
@@ -685,6 +737,35 @@ alterMaterializedViewBody
     : identifier materializedViewRefreshClause
     ;
 
+// ============================================================================
+// DDL STATEMENTS - CONTINUOUS AGGREGATE
+// ============================================================================
+
+/**
+ * CREATE CONTINUOUS AGGREGATE statement
+ * Syntax: CREATE CONTINUOUS AGGREGATE [IF NOT EXISTS] name AS selectStatement
+ */
+createContinuousAggregateBody
+    : (IF NOT EXISTS)? identifier
+      AS selectStatement
+    ;
+
+/**
+ * DROP CONTINUOUS AGGREGATE statement
+ * Syntax: DROP CONTINUOUS AGGREGATE [IF EXISTS] name
+ */
+dropContinuousAggregateBody
+    : (IF EXISTS)? identifier
+    ;
+
+/**
+ * REFRESH CONTINUOUS AGGREGATE statement
+ * Syntax: REFRESH CONTINUOUS AGGREGATE name
+ */
+refreshContinuousAggregateBody
+    : identifier
+    ;
+
 // ============================================================================
 // DDL STATEMENTS - TRUNCATE
 // ============================================================================
@@ -1315,6 +1396,20 @@ identifier
     | MANUAL
     | INCREMENTAL
     | MATERIALIZED
+    | CONTINUOUS
+    | AGGREGATE
+    | TIMESERIES
+    | TAGS
+    | FIELDS
+    | RETENTION
+    | COMPACTION_INTERVAL
+    | SHARDS
+    | DAYS
+    | HOURS
+    | MINUTES
+    | DOWNSAMPLING
+    | POLICY
+    | GRANULARITY
     // Additional keywords allowed as identifiers (matching JavaCC parser)
     | PROPERTY
     | BUCKETS

engine/src/main/java/com/arcadedb/database/LocalDatabase.java

@@ -36,6 +36,7 @@
 import com.arcadedb.engine.WALFile;
 import com.arcadedb.engine.WALFileFactory;
 import com.arcadedb.engine.WALFileFactoryEmbedded;
+import com.arcadedb.engine.timeseries.TimeSeriesBucket;
 import com.arcadedb.exception.ArcadeDBException;
 import com.arcadedb.exception.CommandExecutionException;
 import com.arcadedb.exception.DatabaseIsClosedException;
@@ -74,6 +75,7 @@
 import com.arcadedb.schema.EdgeType;
 import com.arcadedb.schema.LocalDocumentType;
 import com.arcadedb.schema.LocalSchema;
+import com.arcadedb.schema.LocalTimeSeriesType;
 import com.arcadedb.schema.LocalVertexType;
 import com.arcadedb.schema.Property;
 import com.arcadedb.schema.Schema;
@@ -136,7 +138,8 @@ public class LocalDatabase extends RWLockContext implements DatabaseInternal {
       LSMTreeIndexCompacted.NOTUNIQUE_INDEX_EXT,
       LSMTreeIndexCompacted.UNIQUE_INDEX_EXT,
       LSMVectorIndex.FILE_EXT,
-      LSMVectorIndexGraphFile.FILE_EXT);
+      LSMVectorIndexGraphFile.FILE_EXT,
+      TimeSeriesBucket.BUCKET_EXT);
   public final         AtomicLong                                indexCompactions                     =
       new AtomicLong();
   protected final      String                                    name;
@@ -553,6 +556,15 @@ public long countType(final String typeName, final boolean polymorphic) {
     return (Long) executeInReadLock((Callable<Object>) () -> {
       final DocumentType type = schema.getType(typeName);
 
+      // TimeSeries types store data in their own engine, not in regular buckets
+      if (type instanceof LocalTimeSeriesType tsType) {
+        try {
+          return tsType.getEngine().countSamples();
+        } catch (final IOException e) {
+          throw new DatabaseOperationException("Error counting TimeSeries samples for type '" + typeName + "'", e);
+        }
+      }
+
       long total = 0;
       for (final Bucket b : type.getBuckets(polymorphic))
         total += b.count();
@@ -1248,7 +1260,7 @@ public MutableDocument newDocument(final String typeName) {
       throw new IllegalArgumentException("Type is null");
 
     final LocalDocumentType type = schema.getType(typeName);
-    if (!type.getClass().equals(LocalDocumentType.class))
+    if (!type.getClass().equals(LocalDocumentType.class) && !(type instanceof com.arcadedb.schema.LocalTimeSeriesType))
       throw new IllegalArgumentException("Cannot create a document of type '" + typeName + "' because is not a " +
           "document type");
 

engine/src/main/java/com/arcadedb/database/async/DatabaseAsyncAppendSamples.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright © 2021-present Arcade Data Ltd (info@arcadedata.com)
+ *
+ * Licensed 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.
+ *
+ * SPDX-FileCopyrightText: 2021-present Arcade Data Ltd (info@arcadedata.com)
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package com.arcadedb.database.async;
+
+import com.arcadedb.database.DatabaseInternal;
+import com.arcadedb.engine.timeseries.TimeSeriesEngine;
+import com.arcadedb.exception.DatabaseOperationException;
+import com.arcadedb.log.LogManager;
+
+import java.util.logging.Level;
+
+public class DatabaseAsyncAppendSamples implements DatabaseAsyncTask {
+  private final TimeSeriesEngine engine;
+  private final int              shardIndex;
+  private final long[]           timestamps;
+  private final Object[][]       columnValues;
+
+  public DatabaseAsyncAppendSamples(final TimeSeriesEngine engine, final int shardIndex, final long[] timestamps,
+      final Object[][] columnValues) {
+    this.engine = engine;
+    this.shardIndex = shardIndex;
+    this.timestamps = timestamps.clone();
+    this.columnValues = new Object[columnValues.length][];
+    for (int i = 0; i < columnValues.length; i++)
+      this.columnValues[i] = columnValues[i] != null ? columnValues[i].clone() : null;
+  }
+
+  @Override
+  public void execute(final DatabaseAsyncExecutorImpl.AsyncThread async, final DatabaseInternal database) {
+    try {
+      engine.getShard(shardIndex).appendSamples(timestamps, columnValues);
+    } catch (final Exception e) {
+      LogManager.instance().log(this, Level.SEVERE,
+          "Error appending timeseries samples to shard %d of type '%s' (%d points)",
+          e, shardIndex, engine.getTypeName(), timestamps.length);
+      throw new DatabaseOperationException("Error appending timeseries samples to shard " + shardIndex, e);
+    }
+  }
+
+  @Override
+  public String toString() {
+    return "AppendSamples(type=" + engine.getTypeName() + " shard=" + shardIndex + " points=" + timestamps.length + ")";
+  }
+}

engine/src/main/java/com/arcadedb/database/async/DatabaseAsyncExecutor.java

@@ -304,6 +304,16 @@ void newEdgeByKeys(String sourceVertexType, String[] sourceVertexKeyNames, Objec
       boolean bidirectional,
       boolean light, NewEdgeCallback callback, Object... properties);
 
+  /**
+   * Schedules the asynchronous append of time-series samples. The samples are routed to shards in a round-robin
+   * fashion, with each shard pinned to a dedicated async slot for zero-contention parallel ingestion.
+   *
+   * @param typeName     The name of the TimeSeries type
+   * @param timestamps   Array of timestamps for each sample
+   * @param columnValues One array per column (tags + fields), each with the same length as timestamps
+   */
+  void appendSamples(String typeName, long[] timestamps, Object[]... columnValues);
+
   /**
    * Forces the shutdown of the asynchronous threads.
    */

engine/src/main/java/com/arcadedb/database/async/DatabaseAsyncExecutorImpl.java

@@ -31,12 +31,14 @@
 import com.arcadedb.engine.Bucket;
 import com.arcadedb.engine.ErrorRecordCallback;
 import com.arcadedb.engine.WALFile;
+import com.arcadedb.engine.timeseries.TimeSeriesEngine;
 import com.arcadedb.exception.DatabaseOperationException;
 import com.arcadedb.graph.Vertex;
 import com.arcadedb.index.IndexInternal;
 import com.arcadedb.log.LogManager;
 import com.arcadedb.schema.DocumentType;
 import com.arcadedb.schema.EdgeType;
+import com.arcadedb.schema.LocalTimeSeriesType;
 import com.conversantmedia.util.concurrent.PushPullBlockingQueue;
 
 import java.util.Arrays;
@@ -64,6 +66,7 @@ public class DatabaseAsyncExecutorImpl implements DatabaseAsyncExecutor {
   private       long                 checkForStalledQueuesMaxDelay = 5_000;
   private final AtomicLong           transactionCounter            = new AtomicLong();
   private final AtomicLong           commandRoundRobinIndex        = new AtomicLong();
+  private final AtomicLong           tsAppendCounter               = new AtomicLong();
 
   // SPECIAL TASKS
   public final static DatabaseAsyncTask FORCE_EXIT = new DatabaseAsyncTask() {
@@ -643,6 +646,16 @@ public void newEdgeByKeys(final String sourceVertexType, final String[] sourceVe
       newEdge(sourceRID.asVertex(true), edgeType, destinationRID, lightWeight, callback, properties);
   }
 
+  @Override
+  public void appendSamples(final String typeName, final long[] timestamps, final Object[]... columnValues) {
+    final LocalTimeSeriesType tsType = (LocalTimeSeriesType) database.getSchema().getType(typeName);
+    final TimeSeriesEngine engine = tsType.getEngine();
+    final int shardIdx = (int) (tsAppendCounter.getAndIncrement() % engine.getShardCount());
+    final int slot = getSlot(shardIdx);
+    scheduleTask(slot, new DatabaseAsyncAppendSamples(engine, shardIdx, timestamps, columnValues), true,
+        backPressurePercentage);
+  }
+
   /**
    * Test only API.
    */