Improve Time-Series Bucketing Scalability

driver-core/src/main/com/mongodb/client/model/TimeSeriesOptions.java

@@ -18,6 +18,9 @@
 
 import com.mongodb.lang.Nullable;
 
+import java.util.concurrent.TimeUnit;
+
+import static com.mongodb.assertions.Assertions.isTrueArgument;
 import static com.mongodb.assertions.Assertions.notNull;
 
 /**
@@ -31,6 +34,8 @@ public final class TimeSeriesOptions {
     private final String timeField;
     private String metaField;
     private TimeSeriesGranularity granularity;
+    private Long bucketMaxSpanSeconds;
+    private Long bucketRoundingSeconds;
 
     /**
      * Construct a new instance.
@@ -104,12 +109,102 @@ public TimeSeriesOptions granularity(@Nullable final TimeSeriesGranularity granu
         return this;
     }
 
+    /**
+     * Returns the maximum time span between measurements in a bucket.
+     *
+     * @param timeUnit the time unit.
+     * @return time span between measurements.
+     * @since 4.10
+     * @mongodb.server.release 6.3
+     * @mongodb.driver.manual core/timeseries-collections/ Time-series collections
+     * @see #bucketMaxSpan(Long, TimeUnit)
+     */
+    @Nullable
+    public Long getBucketMaxSpan(final TimeUnit timeUnit) {
+        if (bucketMaxSpanSeconds == null) {
+            return null;
+        }
+        return timeUnit.convert(bucketMaxSpanSeconds, TimeUnit.SECONDS);
+    }
+
+    /**
+     * Sets the maximum time span between measurements in a bucket.
+     * <p>
+     * The value of {@code bucketMaxSpanSeconds} must be the same as {@code bucketRoundingSeconds}.
+     * If you set the {@code bucketMaxSpanSeconds}, parameter, you can't set the granularity parameter.
+     * </p>
+     *
+     * @param bucketMaxSpan time span between measurements. After conversion to seconds using {@link TimeUnit#convert(long, java.util.concurrent.TimeUnit)},
+     * the value must be &gt;= 1.
+     * @param timeUnit the time unit.
+     * @return this
+     * @since 4.10
+     * @mongodb.server.release 6.3
+     * @mongodb.driver.manual core/timeseries-collections/ Time-series collections
+     * @see #getBucketMaxSpan(TimeUnit)
+     */
+    public TimeSeriesOptions bucketMaxSpan(@Nullable final Long bucketMaxSpan, final TimeUnit timeUnit) {
+        if (bucketMaxSpan == null) {
+            this.bucketMaxSpanSeconds = null;
+        } else {
+            this.bucketMaxSpanSeconds = TimeUnit.SECONDS.convert(bucketMaxSpan, timeUnit);
+            isTrueArgument("bucketMaxSpan, after conversion to seconds, must be >= 1", bucketMaxSpanSeconds > 0);
+        }
+        return this;
+    }
+
+    /**
+     * Returns the time interval that determines the starting timestamp for a new bucket.
+     *
+     * @param timeUnit the time unit.
+     * @return the time interval.
+     * @since 4.10
+     * @mongodb.server.release 6.3
+     * @mongodb.driver.manual core/timeseries-collections/ Time-series collections
+     * @see #bucketRounding(Long, TimeUnit)
+     */
+    @Nullable
+    public Long getBucketRounding(final TimeUnit timeUnit) {
+        if (bucketRoundingSeconds == null) {
+            return null;
+        }
+        return timeUnit.convert(bucketRoundingSeconds, TimeUnit.SECONDS);
+    }
+
+    /**
+     * Specifies the time interval that determines the starting timestamp for a new bucket.
+     * <p>
+     * The value of {@code bucketRoundingSeconds} must be the same as {@code bucketMaxSpanSeconds}.
+     * If you set the {@code bucketRoundingSeconds}, parameter, you can't set the granularity parameter.
+     * </p>
+     *
+     * @param bucketRounding time interval. After conversion to seconds using {@link TimeUnit#convert(long, java.util.concurrent.TimeUnit)},
+     * the value must be &gt;= 1.
+     * @param timeUnit the time unit.
+     * @return this
+     * @since 4.10
+     * @mongodb.server.release 6.3
+     * @mongodb.driver.manual core/timeseries-collections/ Time-series collections
+     * @see #getBucketRounding(TimeUnit)
+     */
+    public TimeSeriesOptions bucketRounding(@Nullable final Long bucketRounding, final TimeUnit timeUnit) {
+        if (bucketRounding == null) {
+            this.bucketRoundingSeconds = null;
+        } else {
+            this.bucketRoundingSeconds = TimeUnit.SECONDS.convert(bucketRounding, timeUnit);
+            isTrueArgument("bucketRounding, after conversion to seconds, must be >= 1", bucketMaxSpanSeconds > 0);
+        }
+        return this;
+    }
+
     @Override
     public String toString() {
         return "TimeSeriesOptions{"
                 + "timeField='" + timeField + '\''
                 + ", metaField='" + metaField + '\''
                 + ", granularity=" + granularity
+                + ", bucketMaxSpanSeconds=" + bucketMaxSpanSeconds
+                + ", bucketRoundingSeconds=" + bucketRoundingSeconds
                 + '}';
     }
 }

driver-core/src/main/com/mongodb/internal/operation/CreateCollectionOperation.java

@@ -34,11 +34,13 @@
 import org.bson.BsonArray;
 import org.bson.BsonBoolean;
 import org.bson.BsonDocument;
+import org.bson.BsonInt64;
 import org.bson.BsonString;
 
 import java.util.ArrayDeque;
 import java.util.Deque;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
 import static com.mongodb.assertions.Assertions.notNull;
@@ -347,6 +349,14 @@ private BsonDocument getCreateCollectionCommand() {
             if (granularity != null) {
                 timeSeriesDocument.put("granularity", new BsonString(getGranularityAsString(granularity)));
             }
+            Long bucketMaxSpan = timeSeriesOptions.getBucketMaxSpan(TimeUnit.SECONDS);
+            if (bucketMaxSpan != null){
+                timeSeriesDocument.put("bucketMaxSpanSeconds", new BsonInt64(bucketMaxSpan));
+            }
+            Long bucketRounding = timeSeriesOptions.getBucketRounding(TimeUnit.SECONDS);
+            if (bucketRounding != null){
+                timeSeriesDocument.put("bucketRoundingSeconds", new BsonInt64(bucketRounding));
+            }
             document.put("timeseries", timeSeriesDocument);
         }
         if (changeStreamPreAndPostImagesOptions != null) {

driver-core/src/test/resources/unified-test-format/collection-management/timeseries-collection.json

@@ -250,6 +250,71 @@
           ]
         }
       ]
+    },
+    {
+      "description": "createCollection with bucketing options",
+      "runOnRequirements": [
+        {
+          "minServerVersion": "7.0"
+        }
+      ],
+      "operations": [
+        {
+          "name": "dropCollection",
+          "object": "database0",
+          "arguments": {
+            "collection": "test"
+          }
+        },
+        {
+          "name": "createCollection",
+          "object": "database0",
+          "arguments": {
+            "collection": "test",
+            "timeseries": {
+              "timeField": "time",
+              "bucketMaxSpanSeconds": 3600,
+              "bucketRoundingSeconds": 3600
+            }
+          }
+        },
+        {
+          "name": "assertCollectionExists",
+          "object": "testRunner",
+          "arguments": {
+            "databaseName": "ts-tests",
+            "collectionName": "test"
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "command": {
+                  "drop": "test"
+                },
+                "databaseName": "ts-tests"
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "command": {
+                  "create": "test",
+                  "timeseries": {
+                    "timeField": "time",
+                    "bucketMaxSpanSeconds": 3600,
+                    "bucketRoundingSeconds": 3600
+                  }
+                },
+                "databaseName": "ts-tests"
+              }
+            }
+          ]
+        }
+      ]
     }
   ]
 }

driver-kotlin-coroutine/src/test/kotlin/com/mongodb/kotlin/client/coroutine/ExtensionMethodsTest.kt

@@ -35,7 +35,8 @@ class ExtensionMethodsTest {
                 "FindOneAndReplaceOptions",
                 "FindOneAndUpdateOptions",
                 "IndexOptions",
-                "TransactionOptions")
+                "TransactionOptions",
+                "TimeSeriesOptions")
 
         ClassGraph().enableClassInfo().enableMethodInfo().acceptPackages("com.mongodb").scan().use { scanResult ->
             val optionsClassesWithTimeUnit =

driver-kotlin-sync/src/test/kotlin/com/mongodb/kotlin/client/ExtensionMethodsTest.kt

@@ -35,7 +35,8 @@ class ExtensionMethodsTest {
                 "FindOneAndReplaceOptions",
                 "FindOneAndUpdateOptions",
                 "IndexOptions",
-                "TransactionOptions")
+                "TransactionOptions",
+                "TimeSeriesOptions")
 
         ClassGraph().enableClassInfo().enableMethodInfo().acceptPackages("com.mongodb").scan().use { scanResult ->
             val optionsClassesWithTimeUnit =

driver-sync/src/test/functional/com/mongodb/client/unified/UnifiedCrudHelper.java

@@ -1202,6 +1202,12 @@ private TimeSeriesOptions createTimeSeriesOptions(final BsonDocument timeSeriesD
                 case "metaField":
                     options.metaField(cur.getValue().asString().getValue());
                     break;
+                case "bucketMaxSpanSeconds":
+                    options.bucketMaxSpan(cur.getValue().asInt32().longValue(), TimeUnit.SECONDS);
+                    break;
+                case "bucketRoundingSeconds":
+                    options.bucketRounding(cur.getValue().asInt32().longValue(), TimeUnit.SECONDS);
+                    break;
                 case "granularity":
                     options.granularity(createTimeSeriesGranularity(cur.getValue().asString().getValue()));
                     break;