diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md
index 318d5e4da..9a5e0bb62 100644
--- a/CONTRIBUTORS.md
+++ b/CONTRIBUTORS.md
@@ -11,3 +11,4 @@ This file recognizes the people who have made contributions to SQRL.
 | Etienne Chauchot    | echauchot          |
 | Tianchen Wu         | wutianchen         |
 | Brad Bailey         | bradjbailey        |
+| Wellington Mafra    | WellMafra          |
diff --git a/documentation/docs/configuration-engine/postgres.md b/documentation/docs/configuration-engine/postgres.md
index bc72e3e07..9dbae4e0e 100644
--- a/documentation/docs/configuration-engine/postgres.md
+++ b/documentation/docs/configuration-engine/postgres.md
@@ -6,12 +6,17 @@ PostgreSQL is a realtime database that stores the materialized views and tables
 
 No mandatory configuration keys are required. Physical DDL (tables, indexes, views) is produced automatically by the DataSQRL compiler.
 
+| Key                     | Type        | Default | Description                                                                                                    |
+|-------------------------|-------------|---------|----------------------------------------------------------------------------------------------------------------|
+| `partition-ttl-divisor` | **integer** | `100`   | Controls the number of partitions for range-partitioned tables with a TTL (see [Partitioning](#partitioning)). |
+
 ## Basic Configuration
 
 ```json
 {
   "engines": {
     "postgres": {
+      "partition-ttl-divisor": 100,
       "config": {
         // Optional PostgreSQL-specific settings
       }
@@ -20,6 +25,16 @@ No mandatory configuration keys are required. Physical DDL (tables, indexes, vie
 }
 ```
 
+## Partitioning
+
+Tables annotated with `partition_key` on a timestamp column and a `ttl` hint are range-partitioned with pg_partman, and expired partitions are dropped automatically. The partition width is derived from the TTL:
+
+- The TTL duration divided by `partition-ttl-divisor` caps the number of partitions.
+- The unit the TTL is declared with sets the minimum width: `ttl(14 days)` never produces partitions smaller than 1 day, while `ttl(336 hours)` allows hourly partitions.
+- The result is snapped down to the closest calendar-aligned width from: 15 min, 30 min, 1, 2, 4, 6, 8, 12 hours, 1, 2, 4 days, 1, 2, 4, 8, 12 weeks.
+
+For example, `ttl(14 days)` with the default divisor produces 1-day partitions.
+
 ## Cloud Deployment
 
 For cloud deployment configuration (instance sizes, replica counts), see [Cloud Deployment Configuration](cloud-deployment.md#postgresql-enginespostgresdeployment).
diff --git a/documentation/docs/sqrl-language.md b/documentation/docs/sqrl-language.md
index 219cbd078..cc4cae073 100644
--- a/documentation/docs/sqrl-language.md
+++ b/documentation/docs/sqrl-language.md
@@ -329,7 +329,7 @@ Hints live in a `/*+ ... */` comment placed **immediately before** the definitio
 | **query_by_any**            | `query_by_any(col, ...)`                                                   | table          | generate interface with *optional* filter arguments for all listed columns                                                                                                         |
 | **no_query**                | `no_query`                                                                 | table          | hide from interface                                                                                                                                                                |
 | **insert**                  | `insert(type)`                                                             | table          | controls the way how mutations will be written to their target sink. `type` ∈ `SINGLE` (default), `BATCH`, `TRANSACTION`                                                           |
-| **ttl**                     | `ttl(duration)`                                                            | table          | specifies how long the records for this table are retained in the underlying data system before it can be discarded. Expects a duration string like `5 week`. Disabled by default. |
+| **ttl**                     | `ttl(duration)`                                                            | table          | specifies how long the records for this table are retained in the underlying data system before it can be discarded. Expects a duration string like `5 week` with a unit between minute and week (e.g. `30 min`, `36 hours`, `14 days`, `2 weeks`). For range-partitioned tables, the partition width is derived from the duration and its unit (see the [postgres engine configuration](configuration-engine/postgres#partitioning)). Disabled by default. |
 | **cache**                   | `cache(duration)`                                                          | table          | how long the results retrieved from this table can be cached on the server before they are refreshed. Expects a duration string like `10 seconds`. Disabled by default.            |
 | **filtered_distinct_order** | flag                                                                       | DISTINCT table | eliminate updates on order column only before dedup                                                                                                                                |
 | **engine**                  | `enigne(engine_id)`                                                        | table          | pin execution engine (`process`, `database`, `flink`, ...)                                                                                                                         |
diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactory.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactory.java
index f85b73c33..1ab3df510 100644
--- a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactory.java
+++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactory.java
@@ -39,6 +39,7 @@
 import com.datasqrl.sql.DatabaseTypeExtension;
 import com.datasqrl.util.CalciteUtil;
 import java.time.Duration;
+import java.time.temporal.ChronoUnit;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
@@ -167,13 +168,10 @@ public JdbcStatement createTable(JdbcEngineCreateTable createTable) {
     // DagPlanner validates that partition keys are part of primary key
 
     PartitionType partitionType = getPartitionType(createTable, partitionKeys);
-    Duration ttl =
-        createTable
-            .tableAnalysis()
-            .getHints()
-            .getHint(TtlHint.class)
-            .flatMap(TtlHint::getTtl)
-            .orElse(Duration.ZERO);
+    var ttlHint = createTable.tableAnalysis().getHints().getHint(TtlHint.class);
+    Duration ttl = ttlHint.flatMap(TtlHint::getTtl).orElse(Duration.ZERO);
+    ChronoUnit ttlUnit = ttlHint.flatMap(TtlHint::getTtlUnit).orElse(null);
+    String partitionInterval = derivePartitionInterval(partitionType, ttl, ttlUnit).orElse(null);
 
     return new CreateTableJdbcStatement(
         tableName,
@@ -187,6 +185,7 @@ public JdbcStatement createTable(JdbcEngineCreateTable createTable) {
         partitionType,
         partitionType == PartitionType.NONE ? 0 : 1,
         ttl,
+        partitionInterval,
         createTable,
         createTableDdlFactory);
   }
@@ -196,6 +195,15 @@ protected PartitionType getPartitionType(
     return partitionKey.isEmpty() ? PartitionType.NONE : PartitionType.LIST;
   }
 
+  /**
+   * The width of the partitions for range-partitioned tables with a TTL - empty when the dialect
+   * does not support time-based partitioning.
+   */
+  protected Optional<String> derivePartitionInterval(
+      PartitionType partitionType, Duration ttl, ChronoUnit ttlUnit) {
+    return Optional.empty();
+  }
+
   protected List<Field> getColumns(
       List<RelDataTypeField> fields, PlannerHints hints, Documentation documentation) {
     return fields.stream()
diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/CreateTableJdbcStatement.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/CreateTableJdbcStatement.java
index 772a01b74..20d29a7e7 100644
--- a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/CreateTableJdbcStatement.java
+++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/CreateTableJdbcStatement.java
@@ -63,6 +63,9 @@ public class CreateTableJdbcStatement implements JdbcStatement {
   /** The time-to-live of records in this table - ZERO to disable */
   Duration ttl;
 
+  /** Partition width for range-partitioned tables, derived from the ttl - null when not ranged */
+  String partitionInterval;
+
   /** The engine table - nullable and not set when deserialized */
   @JsonIgnore JdbcEngineCreateTable engineTable;
 
@@ -78,7 +81,8 @@ public CreateTableJdbcStatement(
       @JsonProperty("partitionKey") List<String> partitionKey,
       @JsonProperty("partitionType") PartitionType partitionType,
       @JsonProperty("numPartitions") int numPartitions,
-      @JsonProperty("ttl") Duration ttl) {
+      @JsonProperty("ttl") Duration ttl,
+      @JsonProperty("partitionInterval") String partitionInterval) {
     this.name = name;
     this.description = description;
     this.fields = fields;
@@ -87,6 +91,7 @@ public CreateTableJdbcStatement(
     this.partitionType = partitionType;
     this.numPartitions = numPartitions;
     this.ttl = ttl;
+    this.partitionInterval = partitionInterval;
     this.engineTable = null;
     this.ddlFactory = null;
   }
diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresJdbcEngine.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresJdbcEngine.java
index 1775e469e..c9abacb9a 100644
--- a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresJdbcEngine.java
+++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresJdbcEngine.java
@@ -60,7 +60,7 @@ protected DatabaseType getDatabaseType() {
 
   @Override
   public JdbcStatementFactory getStatementFactory() {
-    return new PostgresStatementFactory();
+    return new PostgresStatementFactory(engineConfig);
   }
 
   @Override
diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresPartitionInterval.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresPartitionInterval.java
new file mode 100644
index 000000000..0e37958ec
--- /dev/null
+++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresPartitionInterval.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright © 2021 DataSQRL (contact@datasqrl.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.
+ */
+package com.datasqrl.engine.database.relational;
+
+import com.google.common.collect.ImmutableMap;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.Map;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+/** Computes the pg_partman partition width for a range-partitioned table with a TTL. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public final class PostgresPartitionInterval {
+
+  /** Calendar-aligned partition widths pg_partman can use, keyed by width in minutes. */
+  private static final Map<Integer, String> PARTITION_MENU =
+      ImmutableMap.<Integer, String>builder()
+          .put(15, "15 minutes")
+          .put(30, "30 minutes")
+          .put(60, "1 hour")
+          .put(120, "2 hours")
+          .put(240, "4 hours")
+          .put(360, "6 hours")
+          .put(480, "8 hours")
+          .put(720, "12 hours")
+          .put(1440, "1 day")
+          .put(2880, "2 days")
+          .put(5760, "4 days")
+          .put(10080, "1 week")
+          .put(20160, "2 weeks")
+          .put(40320, "4 weeks")
+          .put(80640, "8 weeks")
+          .put(120960, "12 weeks")
+          .buildOrThrow();
+
+  /**
+   * Picks the partition width for the given TTL: the TTL divided by the divisor caps the partition
+   * count, while the unit the TTL was declared with sets the floor. The result is snapped down to
+   * the closest calendar-aligned width from the menu.
+   */
+  public static String asString(Duration ttl, ChronoUnit ttlUnit, int partitionTtlDivisor) {
+    var floorMinutes = ttlUnit.getDuration().toMinutes();
+    var targetMinutes = Math.max(ttl.toMinutes() / (double) partitionTtlDivisor, floorMinutes);
+
+    String interval = null;
+    for (var entry : PARTITION_MENU.entrySet()) {
+      if (interval == null || entry.getKey() <= targetMinutes) {
+        interval = entry.getValue();
+      }
+    }
+
+    return interval;
+  }
+}
diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresStatementFactory.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresStatementFactory.java
index 77fb171c4..8e2b3e530 100644
--- a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresStatementFactory.java
+++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresStatementFactory.java
@@ -20,6 +20,7 @@
 import com.datasqrl.calcite.Dialect;
 import com.datasqrl.calcite.dialect.ExtendedPostgresSqlDialect;
 import com.datasqrl.config.JdbcDialect;
+import com.datasqrl.config.PackageJson.EngineConfig;
 import com.datasqrl.deployment.model.JdbcStatementModel.PartitionType;
 import com.datasqrl.deployment.model.JdbcStatementModel.Type;
 import com.datasqrl.engine.database.relational.ddl.CreateIndexDDL;
@@ -34,6 +35,8 @@
 import com.datasqrl.sql.DatabaseTypeExtension;
 import com.datasqrl.util.CalciteUtil;
 import com.datasqrl.util.ServiceLoaderDiscovery;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -46,8 +49,19 @@
 
 public class PostgresStatementFactory extends AbstractJdbcStatementFactory {
 
-  public PostgresStatementFactory() {
+  public static final String PARTITION_TTL_DIVISOR_KEY = "partition-ttl-divisor";
+
+  private final int partitionTtlDivisor;
+
+  public PostgresStatementFactory(EngineConfig engineConfig) {
+    this(Integer.parseInt(engineConfig.getSetting(PARTITION_TTL_DIVISOR_KEY)));
+  }
+
+  public PostgresStatementFactory(int partitionTtlDivisor) {
     super(Dialect.POSTGRES, new PostgresCreateTableDdlFactory(true));
+    checkArgument(
+        partitionTtlDivisor > 0, "%s must be a positive number", PARTITION_TTL_DIVISOR_KEY);
+    this.partitionTtlDivisor = partitionTtlDivisor;
   }
 
   @Override
@@ -92,6 +106,15 @@ protected PartitionType getPartitionType(
     return CalciteUtil.isTimestamp(colType.getType()) ? PartitionType.RANGE : PartitionType.HASH;
   }
 
+  @Override
+  protected Optional<String> derivePartitionInterval(
+      PartitionType partitionType, Duration ttl, ChronoUnit ttlUnit) {
+    if (partitionType != PartitionType.RANGE || ttl == null || ttl.isZero() || ttlUnit == null) {
+      return Optional.empty();
+    }
+    return Optional.of(PostgresPartitionInterval.asString(ttl, ttlUnit, partitionTtlDivisor));
+  }
+
   @Override
   public List<JdbcStatement> applyTableExtensions(Collection<CreateTableJdbcStatement> tables) {
     var res = new ArrayList<JdbcStatement>();
diff --git a/sqrl-planner/src/main/java/com/datasqrl/function/translation/postgres/extensions/PgPartmanExtension.java b/sqrl-planner/src/main/java/com/datasqrl/function/translation/postgres/extensions/PgPartmanExtension.java
index 97415aa38..bd4285370 100644
--- a/sqrl-planner/src/main/java/com/datasqrl/function/translation/postgres/extensions/PgPartmanExtension.java
+++ b/sqrl-planner/src/main/java/com/datasqrl/function/translation/postgres/extensions/PgPartmanExtension.java
@@ -19,6 +19,7 @@
 import com.datasqrl.engine.database.relational.CreateTableJdbcStatement;
 import com.datasqrl.sql.DatabaseTableExtension;
 import com.google.auto.service.AutoService;
+import com.google.common.base.Preconditions;
 import java.time.Duration;
 import java.util.Collection;
 import java.util.Comparator;
@@ -60,6 +61,11 @@ private void appendTableDdl(StringBuilder sb, CreateTableJdbcStatement createTab
     // 'public."Readings"' never matches relname 'Readings' and create_parent fails.
     var parentTable = "public." + createTable.getName();
     var ttl = createTable.getTtl();
+    var interval =
+        Preconditions.checkNotNull(
+            createTable.getPartitionInterval(),
+            "Missing partition interval for partitioned table %s with TTL",
+            createTable.getName());
 
     // p_type was removed in pg_partman 5.x; the default (range) is what we need
     sb.append(
@@ -72,7 +78,7 @@ private void appendTableDdl(StringBuilder sb, CreateTableJdbcStatement createTab
         );
 
         """
-            .formatted(parentTable, createTable.getPartitionKey().get(0), deriveInterval(ttl)));
+            .formatted(parentTable, createTable.getPartitionKey().get(0), interval));
 
     sb.append(
         """
@@ -92,18 +98,6 @@ private boolean isNotPartmanTable(CreateTableJdbcStatement createTable) {
         || createTable.getPartitionKey().isEmpty();
   }
 
-  /** Picks a partition interval that keeps the partition count reasonable for the given TTL. */
-  static String deriveInterval(Duration ttl) {
-    long seconds = ttl.getSeconds();
-    if (seconds < 7 * 24 * 3600) {
-      return "1 day";
-    } else if (seconds < 90L * 24 * 3600) {
-      return "1 week";
-    } else {
-      return "1 month";
-    }
-  }
-
   static String formatRetention(Duration ttl) {
     long days = ttl.toDays();
     if (days > 0) {
diff --git a/sqrl-planner/src/main/java/com/datasqrl/planner/hint/CacheHint.java b/sqrl-planner/src/main/java/com/datasqrl/planner/hint/CacheHint.java
index e22536010..dc1a01de8 100644
--- a/sqrl-planner/src/main/java/com/datasqrl/planner/hint/CacheHint.java
+++ b/sqrl-planner/src/main/java/com/datasqrl/planner/hint/CacheHint.java
@@ -15,10 +15,11 @@
  */
 package com.datasqrl.planner.hint;
 
-import static com.datasqrl.planner.hint.TtlHint.parseDuration;
-
+import com.datasqrl.error.ErrorLabel;
 import com.datasqrl.planner.parser.ParsedObject;
 import com.datasqrl.planner.parser.SqrlHint;
+import com.datasqrl.planner.parser.StatementParserException;
+import com.datasqrl.util.TimeUtils;
 import com.google.auto.service.AutoService;
 import java.time.Duration;
 import lombok.Getter;
@@ -48,4 +49,29 @@ public String getName() {
       return HINT_NAME;
     }
   }
+
+  private static Duration parseDuration(ParsedObject<SqrlHint> source) {
+    var arguments = source.get().options();
+    if (arguments == null || arguments.isEmpty()) {
+      return null;
+    }
+    if (arguments.size() != 1 || arguments.get(0) == null) {
+      throw new StatementParserException(
+          ErrorLabel.GENERIC,
+          source.getFileLocation(),
+          "%s hint only supports one duration argument (e.g. `2 days`).",
+          source.get().name());
+    }
+    try {
+      return TimeUtils.parseDuration(arguments.get(0));
+    } catch (Exception e) {
+      throw new StatementParserException(
+          ErrorLabel.GENERIC,
+          source.getFileLocation(),
+          "%s hint does not have a valid duration argument: %s. Expected `2 days` or `10 s`. "
+              + e.getMessage(),
+          source.get().name(),
+          arguments.get(0));
+    }
+  }
 }
diff --git a/sqrl-planner/src/main/java/com/datasqrl/planner/hint/TtlHint.java b/sqrl-planner/src/main/java/com/datasqrl/planner/hint/TtlHint.java
index b8fa3cf4e..f7eda98be 100644
--- a/sqrl-planner/src/main/java/com/datasqrl/planner/hint/TtlHint.java
+++ b/sqrl-planner/src/main/java/com/datasqrl/planner/hint/TtlHint.java
@@ -19,32 +19,51 @@
 import com.datasqrl.planner.parser.ParsedObject;
 import com.datasqrl.planner.parser.SqrlHint;
 import com.datasqrl.planner.parser.StatementParserException;
+import com.datasqrl.util.TimeUtils;
 import com.google.auto.service.AutoService;
 import java.time.Duration;
+import java.time.temporal.ChronoUnit;
 import java.util.Optional;
-import org.apache.flink.util.TimeUtils;
 
 public class TtlHint extends PlannerHint {
 
   public static final String HINT_NAME = "ttl";
 
   private final Duration ttl;
+  private final ChronoUnit ttlUnit;
 
-  protected TtlHint(ParsedObject<SqrlHint> source, Duration ttlDuration) {
+  protected TtlHint(ParsedObject<SqrlHint> source, Duration ttl, ChronoUnit ttlUnit) {
     super(source, Type.DAG);
-    this.ttl = ttlDuration;
+    this.ttl = ttl;
+    this.ttlUnit = ttlUnit;
   }
 
   public Optional<Duration> getTtl() {
     return Optional.ofNullable(ttl);
   }
 
+  /** The unit the TTL was declared with - determines the smallest allowed partition width */
+  public Optional<ChronoUnit> getTtlUnit() {
+    return Optional.ofNullable(ttlUnit);
+  }
+
   @AutoService(Factory.class)
   public static class TtlHintFactory implements Factory {
 
     @Override
     public PlannerHint create(ParsedObject<SqrlHint> source) {
-      return new TtlHint(source, parseDuration(source));
+      var arguments = source.get().options();
+      if (arguments == null || arguments.isEmpty()) {
+        return new TtlHint(source, null, null);
+      }
+      if (arguments.size() != 1 || arguments.get(0) == null) {
+        throw new StatementParserException(
+            ErrorLabel.GENERIC,
+            source.getFileLocation(),
+            "%s hint only supports one duration argument (e.g. `2 days`).",
+            source.get().name());
+      }
+      return parseTtlArgument(source, arguments.get(0).trim());
     }
 
     @Override
@@ -53,28 +72,26 @@ public String getName() {
     }
   }
 
-  public static Duration parseDuration(ParsedObject<SqrlHint> source) {
-    var arguments = source.get().options();
-    if (arguments == null || arguments.isEmpty()) {
-      return null;
-    }
-    if (arguments.size() != 1 || arguments.get(0) == null) {
-      throw new StatementParserException(
-          ErrorLabel.GENERIC,
-          source.getFileLocation(),
-          "%s hint only supports one duration argument (e.g. `2 days`).",
-          source.get().name());
-    }
+  private static TtlHint parseTtlArgument(ParsedObject<SqrlHint> source, String argument) {
+    Duration ttl = null;
+    ChronoUnit unit = null;
     try {
-      return TimeUtils.parseDuration(arguments.get(0));
+      ttl = TimeUtils.parseDuration(argument);
+      unit = TimeUtils.parseDurationUnit(argument);
     } catch (Exception e) {
+      // fall through to the shared error below
+    }
+    if (unit == null
+        || unit.compareTo(ChronoUnit.MINUTES) < 0
+        || unit.compareTo(ChronoUnit.DAYS) > 0) {
       throw new StatementParserException(
           ErrorLabel.GENERIC,
           source.getFileLocation(),
-          "%s hint does not have a valid duration argument: %s. Expected `2 days` or `10 s`. "
-              + e.getMessage(),
+          "%s hint does not have a valid duration argument: %s. Expected a positive number with a"
+              + " unit between minute and day, e.g. `30 min`, `36 hours`, or `14 days`.",
           source.get().name(),
-          arguments.get(0));
+          argument);
     }
+    return new TtlHint(source, ttl, unit);
   }
 }
diff --git a/sqrl-planner/src/main/java/com/datasqrl/util/TimeUtils.java b/sqrl-planner/src/main/java/com/datasqrl/util/TimeUtils.java
new file mode 100644
index 000000000..e48667471
--- /dev/null
+++ b/sqrl-planner/src/main/java/com/datasqrl/util/TimeUtils.java
@@ -0,0 +1,59 @@
+/*
+ * Copyright © 2021 DataSQRL (contact@datasqrl.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.
+ */
+package com.datasqrl.util;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.Arrays;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+/** Duration parsing utilities built on top of Flink's {@link org.apache.flink.util.TimeUtils}. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public final class TimeUtils {
+
+  /**
+   * Parses the given string to a {@link Duration}, accepting the same formats as Flink's {@link
+   * org.apache.flink.util.TimeUtils#parseDuration(String)}, e.g. {@code 30 min}, {@code 2 days}. If
+   * no unit label is specified, the value is interpreted as milliseconds.
+   */
+  public static Duration parseDuration(String text) {
+    return org.apache.flink.util.TimeUtils.parseDuration(text);
+  }
+
+  /**
+   * Extracts the unit the given duration string was declared with, e.g. {@code 2 days} yields
+   * {@link ChronoUnit#DAYS}. If no unit label is specified, {@link ChronoUnit#MILLIS} is returned,
+   * mirroring {@link #parseDuration(String)}.
+   */
+  public static ChronoUnit parseDurationUnit(String text) {
+    var trimmed = text.trim();
+    var pos = 0;
+    while (pos < trimmed.length() && Character.isDigit(trimmed.charAt(pos))) {
+      pos++;
+    }
+    if (pos == 0) {
+      throw new NumberFormatException("text does not start with a number");
+    }
+    // Flink keeps its label-to-unit map private, so recover the unit by parsing a unit-sized
+    // duration through Flink and matching it back to a ChronoUnit
+    var unitDuration = org.apache.flink.util.TimeUtils.parseDuration("1" + trimmed.substring(pos));
+    return Arrays.stream(ChronoUnit.values())
+        .filter(unit -> unit.getDuration().equals(unitDuration))
+        .findFirst()
+        .orElseThrow(() -> new IllegalArgumentException("Unrecognized time unit in: " + trimmed));
+  }
+}
diff --git a/sqrl-planner/src/main/resources/default-package.json b/sqrl-planner/src/main/resources/default-package.json
index a882ad626..282c979e9 100644
--- a/sqrl-planner/src/main/resources/default-package.json
+++ b/sqrl-planner/src/main/resources/default-package.json
@@ -33,6 +33,9 @@
     "duckdb": {
       "url": "jdbc:duckdb:"
     },
+    "postgres": {
+      "partition-ttl-divisor": 100
+    },
     "kafka": {
       "retention": null,
       "watermark": "0 ms",
diff --git a/sqrl-planner/src/main/resources/jsonSchema/packageSchema.json b/sqrl-planner/src/main/resources/jsonSchema/packageSchema.json
index 5be2a3813..a5c0e3f6d 100644
--- a/sqrl-planner/src/main/resources/jsonSchema/packageSchema.json
+++ b/sqrl-planner/src/main/resources/jsonSchema/packageSchema.json
@@ -265,6 +265,10 @@
         "postgres": {
           "type": "object",
           "properties": {
+            "partition-ttl-divisor": {
+              "description": "Divisor applied to a table's TTL to cap the number of pg_partman partitions.",
+              "type": "integer"
+            },
             "config": {
               "type": "object",
               "minProperties": 1
diff --git a/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/PostgresPartitionIntervalTest.java b/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/PostgresPartitionIntervalTest.java
new file mode 100644
index 000000000..ea5553f29
--- /dev/null
+++ b/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/PostgresPartitionIntervalTest.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright © 2021 DataSQRL (contact@datasqrl.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.
+ */
+package com.datasqrl.engine.database.relational;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+class PostgresPartitionIntervalTest {
+
+  @ParameterizedTest
+  @CsvSource({
+    // TTL unit sets the floor when ttl/divisor is smaller
+    "14, DAYS, 100, 1 day",
+    "30, DAYS, 100, 1 day",
+    "36, HOURS, 100, 1 hour",
+    "90, MINUTES, 100, 15 minutes",
+    // divisor drives the width for long TTLs, snapped down to the menu
+    "84, DAYS, 4, 2 weeks",
+    "28, DAYS, 3, 1 week",
+    "30, DAYS, 10, 2 days",
+    // target below the smallest menu entry falls back to the smallest
+    "30, MINUTES, 100, 15 minutes"
+  })
+  void givenTtlAndDivisor_whenAsString_thenMenuWidth(
+      long amount, ChronoUnit unit, int divisor, String expected) {
+    var ttl = Duration.of(amount, unit);
+
+    assertThat(PostgresPartitionInterval.asString(ttl, unit, divisor)).isEqualTo(expected);
+  }
+}
diff --git a/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/PostgresStatementFactoryTest.java b/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/PostgresStatementFactoryTest.java
new file mode 100644
index 000000000..2666beade
--- /dev/null
+++ b/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/PostgresStatementFactoryTest.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright © 2021 DataSQRL (contact@datasqrl.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.
+ */
+package com.datasqrl.engine.database.relational;
+
+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;
+
+import com.datasqrl.config.PackageJson.EngineConfig;
+import com.datasqrl.deployment.model.JdbcStatementModel.PartitionType;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import org.junit.jupiter.api.Test;
+
+class PostgresStatementFactoryTest {
+
+  @Test
+  void givenNonRangeOrNoTtl_whenDeriveInterval_thenEmpty() {
+    var factory = new PostgresStatementFactory(100);
+
+    assertThat(
+            factory.derivePartitionInterval(
+                PartitionType.HASH, Duration.ofDays(14), ChronoUnit.DAYS))
+        .isEmpty();
+    assertThat(factory.derivePartitionInterval(PartitionType.RANGE, Duration.ZERO, ChronoUnit.DAYS))
+        .isEmpty();
+    assertThat(factory.derivePartitionInterval(PartitionType.RANGE, null, null)).isEmpty();
+    assertThat(factory.derivePartitionInterval(PartitionType.RANGE, Duration.ofDays(14), null))
+        .isEmpty();
+  }
+
+  @Test
+  void givenRangeAndTtl_whenDeriveInterval_thenWidthReturned() {
+    var factory = new PostgresStatementFactory(100);
+
+    assertThat(
+            factory.derivePartitionInterval(
+                PartitionType.RANGE, Duration.ofDays(14), ChronoUnit.DAYS))
+        .contains("1 day");
+  }
+
+  @Test
+  void givenConfiguredDivisor_whenCreate_thenDivisorApplied() {
+    var engineConfig = mock(EngineConfig.class);
+    when(engineConfig.getSetting(PostgresStatementFactory.PARTITION_TTL_DIVISOR_KEY))
+        .thenReturn("4");
+    var factory = new PostgresStatementFactory(engineConfig);
+
+    assertThat(
+            factory.derivePartitionInterval(
+                PartitionType.RANGE, Duration.ofDays(84), ChronoUnit.DAYS))
+        .contains("2 weeks");
+  }
+
+  @Test
+  void givenInvalidDivisor_whenCreate_thenThrows() {
+    var engineConfig = mock(EngineConfig.class);
+    when(engineConfig.getSetting(PostgresStatementFactory.PARTITION_TTL_DIVISOR_KEY))
+        .thenReturn("not-a-number");
+
+    assertThatThrownBy(() -> new PostgresStatementFactory(engineConfig))
+        .isInstanceOf(NumberFormatException.class);
+
+    assertThatThrownBy(() -> new PostgresStatementFactory(0))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("partition-ttl-divisor");
+  }
+}
diff --git a/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/ddl/PostgresCreateTableDdlFactoryTest.java b/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/ddl/PostgresCreateTableDdlFactoryTest.java
index 01ce088f7..2bace60cd 100644
--- a/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/ddl/PostgresCreateTableDdlFactoryTest.java
+++ b/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/ddl/PostgresCreateTableDdlFactoryTest.java
@@ -37,7 +37,8 @@ private static CreateTableJdbcStatement stmt(
         partitionKey,
         partitionType,
         1,
-        ttl);
+        ttl,
+        null);
   }
 
   @Test
diff --git a/sqrl-planner/src/test/java/com/datasqrl/function/translation/postgres/extensions/PgPartmanExtensionTest.java b/sqrl-planner/src/test/java/com/datasqrl/function/translation/postgres/extensions/PgPartmanExtensionTest.java
index 40475bb7b..fd3cca27e 100644
--- a/sqrl-planner/src/test/java/com/datasqrl/function/translation/postgres/extensions/PgPartmanExtensionTest.java
+++ b/sqrl-planner/src/test/java/com/datasqrl/function/translation/postgres/extensions/PgPartmanExtensionTest.java
@@ -29,8 +29,25 @@ class PgPartmanExtensionTest {
 
   private static CreateTableJdbcStatement table(
       String name, PartitionType partitionType, List<String> partitionKey, Duration ttl) {
+    return table(name, partitionType, partitionKey, ttl, "1 day");
+  }
+
+  private static CreateTableJdbcStatement table(
+      String name,
+      PartitionType partitionType,
+      List<String> partitionKey,
+      Duration ttl,
+      String partitionInterval) {
     return new CreateTableJdbcStatement(
-        name, null, List.of(), List.of("id", "time"), partitionKey, partitionType, 1, ttl);
+        name,
+        null,
+        List.of(),
+        List.of("id", "time"),
+        partitionKey,
+        partitionType,
+        1,
+        ttl,
+        partitionInterval);
   }
 
   @Test
@@ -57,7 +74,7 @@ void givenRangeTtlTable_whenGetDdl_thenFullSetupSql() {
         .contains("SELECT partman.create_parent")
         .contains("p_parent_table => 'public.Orders_1'")
         .contains("p_control => 'time'")
-        .contains("p_interval => '1 week'")
+        .contains("p_interval => '1 day'")
         .contains("p_premake => 4")
         .contains("UPDATE partman.part_config")
         .contains("retention = '30 days'")
@@ -83,13 +100,18 @@ void givenMixedTables_whenGetDdl_thenOnlyRangeTtlTablesSortedByName() {
   }
 
   @Test
-  void givenTtl_whenDeriveInterval_thenBucketedByDuration() {
-    assertThat(PgPartmanExtension.deriveInterval(Duration.ofHours(6))).isEqualTo("1 day");
-    assertThat(PgPartmanExtension.deriveInterval(Duration.ofDays(6))).isEqualTo("1 day");
-    assertThat(PgPartmanExtension.deriveInterval(Duration.ofDays(7))).isEqualTo("1 week");
-    assertThat(PgPartmanExtension.deriveInterval(Duration.ofDays(89))).isEqualTo("1 week");
-    assertThat(PgPartmanExtension.deriveInterval(Duration.ofDays(90))).isEqualTo("1 month");
-    assertThat(PgPartmanExtension.deriveInterval(Duration.ofDays(365))).isEqualTo("1 month");
+  void givenPartitionInterval_whenGetDdl_thenUsedAsIs() {
+    assertThat(
+            extension.getDdl(
+                List.of(
+                    table(
+                        "Metrics",
+                        PartitionType.RANGE,
+                        List.of("time"),
+                        Duration.ofDays(14),
+                        "2 weeks"))))
+        .contains("p_interval => '2 weeks'")
+        .contains("retention = '14 days'");
   }
 
   @Test
diff --git a/sqrl-planner/src/test/java/com/datasqrl/planner/PlanModelSerializationTest.java b/sqrl-planner/src/test/java/com/datasqrl/planner/PlanModelSerializationTest.java
index e52867f42..a2ae884e5 100644
--- a/sqrl-planner/src/test/java/com/datasqrl/planner/PlanModelSerializationTest.java
+++ b/sqrl-planner/src/test/java/com/datasqrl/planner/PlanModelSerializationTest.java
@@ -52,6 +52,7 @@ void givenJdbcPhysicalPlan_whenMapped_thenReturnsJdbcPlanModel() {
             4,
             Duration.ofSeconds(30),
             null,
+            null,
             statement -> "CREATE TABLE orders");
     var view =
         new GenericJdbcStatement(
diff --git a/sqrl-planner/src/test/java/com/datasqrl/planner/hint/TtlHintTest.java b/sqrl-planner/src/test/java/com/datasqrl/planner/hint/TtlHintTest.java
new file mode 100644
index 000000000..d83f43fbd
--- /dev/null
+++ b/sqrl-planner/src/test/java/com/datasqrl/planner/hint/TtlHintTest.java
@@ -0,0 +1,148 @@
+/*
+ * Copyright © 2021 DataSQRL (contact@datasqrl.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.
+ */
+package com.datasqrl.planner.hint;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.datasqrl.error.ErrorLocation.FileLocation;
+import com.datasqrl.planner.parser.ParsedObject;
+import com.datasqrl.planner.parser.SqrlHint;
+import com.datasqrl.planner.parser.StatementParserException;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.Arrays;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TtlHintTest {
+
+  private final TtlHint.TtlHintFactory ttlFactory = new TtlHint.TtlHintFactory();
+  private final CacheHint.CacheHintFactory cacheFactory = new CacheHint.CacheHintFactory();
+
+  private static ParsedObject<SqrlHint> hint(String name, String... args) {
+    return new ParsedObject<>(new SqrlHint(name, List.of(args)), FileLocation.START);
+  }
+
+  @Test
+  void givenSingleDurationArg_whenCreate_thenTtlAndUnitSet() {
+    var hint = (TtlHint) ttlFactory.create(hint("ttl", "14 days"));
+
+    assertThat(hint.getTtl()).contains(Duration.ofDays(14));
+    assertThat(hint.getTtlUnit()).contains(ChronoUnit.DAYS);
+  }
+
+  @Test
+  void givenNoArgs_whenCreate_thenEmptyTtlAndUnit() {
+    var hint = (TtlHint) ttlFactory.create(hint("ttl"));
+
+    assertThat(hint.getTtl()).isEmpty();
+    assertThat(hint.getTtlUnit()).isEmpty();
+  }
+
+  @ParameterizedTest
+  @CsvSource({
+    "30 min, 30, MINUTES",
+    "45 minutes, 45, MINUTES",
+    "36 hours, 2160, HOURS",
+    "1 h, 60, HOURS",
+    "14 days, 20160, DAYS",
+    "1 d, 1440, DAYS",
+    "14days, 20160, DAYS"
+  })
+  void givenSupportedUnit_whenCreate_thenTtlAndUnitParsed(
+      String argument, long expectedMinutes, ChronoUnit expectedUnit) {
+    var hint = (TtlHint) ttlFactory.create(hint("ttl", argument));
+
+    assertThat(hint.getTtl()).contains(Duration.ofMinutes(expectedMinutes));
+    assertThat(hint.getTtlUnit()).contains(expectedUnit);
+  }
+
+  @ParameterizedTest
+  @ValueSource(
+      strings = {
+        "10 s",
+        "10 seconds",
+        "500 ms",
+        "2 weeks",
+        "5 week",
+        "3 months",
+        "1 year",
+        "fortnight",
+        "14",
+        "days"
+      })
+  void givenUnsupportedUnitOrFormat_whenCreate_thenThrows(String argument) {
+    assertThatThrownBy(() -> ttlFactory.create(hint("ttl", argument)))
+        .isInstanceOf(StatementParserException.class)
+        .hasMessageContaining("unit between minute and day");
+  }
+
+  @Test
+  void givenTwoArgs_whenCreate_thenThrows() {
+    assertThatThrownBy(() -> ttlFactory.create(hint("ttl", "14 days", "1 day")))
+        .isInstanceOf(StatementParserException.class)
+        .hasMessageContaining("one duration argument");
+  }
+
+  @Test
+  void givenSingleArg_whenCreateCacheHint_thenDurationSet() {
+    var hint = (CacheHint) cacheFactory.create(hint("cache", "5 min"));
+
+    assertThat(hint.getDuration()).isEqualTo(Duration.ofMinutes(5));
+  }
+
+  @Test
+  void givenTwoArgs_whenCreateCacheHint_thenThrows() {
+    assertThatThrownBy(() -> cacheFactory.create(hint("cache", "5 min", "1 day")))
+        .isInstanceOf(StatementParserException.class);
+  }
+
+  @Test
+  void givenNullOptions_whenCreate_thenEmptyTtlAndUnit() {
+    var hint = (TtlHint) ttlFactory.create(nullArgsHint("ttl", (List<String>) null));
+
+    assertThat(hint.getTtl()).isEmpty();
+    assertThat(hint.getTtlUnit()).isEmpty();
+  }
+
+  @Test
+  void givenNullOptions_whenCreateCacheHint_thenZeroDuration() {
+    var hint = (CacheHint) cacheFactory.create(nullArgsHint("cache", (List<String>) null));
+
+    assertThat(hint.getDuration()).isEqualTo(Duration.ZERO);
+  }
+
+  @Test
+  void givenNullFirstArg_whenCreate_thenThrows() {
+    assertThatThrownBy(() -> ttlFactory.create(nullArgsHint("ttl", Arrays.asList((String) null))))
+        .isInstanceOf(StatementParserException.class);
+  }
+
+  @Test
+  void givenNullFirstArg_whenCreateCacheHint_thenThrows() {
+    assertThatThrownBy(
+            () -> cacheFactory.create(nullArgsHint("cache", Arrays.asList((String) null))))
+        .isInstanceOf(StatementParserException.class);
+  }
+
+  private static ParsedObject<SqrlHint> nullArgsHint(String name, List<String> args) {
+    return new ParsedObject<>(new SqrlHint(name, args), FileLocation.START);
+  }
+}
diff --git a/sqrl-planner/src/test/java/com/datasqrl/util/TimeUtilsTest.java b/sqrl-planner/src/test/java/com/datasqrl/util/TimeUtilsTest.java
new file mode 100644
index 000000000..6e3487240
--- /dev/null
+++ b/sqrl-planner/src/test/java/com/datasqrl/util/TimeUtilsTest.java
@@ -0,0 +1,71 @@
+/*
+ * Copyright © 2021 DataSQRL (contact@datasqrl.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.
+ */
+package com.datasqrl.util;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TimeUtilsTest {
+
+  @ParameterizedTest
+  @CsvSource({
+    "30 min, PT30M",
+    "45 minutes, PT45M",
+    "36 hours, PT36H",
+    "1 h, PT1H",
+    "14 days, PT336H",
+    "1 d, PT24H",
+    "14days, PT336H",
+    "10 s, PT10S",
+    "500 ms, PT0.5S",
+    "500, PT0.5S"
+  })
+  void givenDurationString_whenParseDuration_thenParsed(String argument, Duration expected) {
+    assertThat(TimeUtils.parseDuration(argument)).isEqualTo(expected);
+  }
+
+  @ParameterizedTest
+  @CsvSource({
+    "30 min, MINUTES",
+    "45 minutes, MINUTES",
+    "1 m, MINUTES",
+    "36 hours, HOURS",
+    "1 h, HOURS",
+    "14 days, DAYS",
+    "1 d, DAYS",
+    "14days, DAYS",
+    "10 s, SECONDS",
+    "500 ms, MILLIS",
+    "500, MILLIS"
+  })
+  void givenDurationString_whenParseDurationUnit_thenUnitExtracted(
+      String argument, ChronoUnit expected) {
+    assertThat(TimeUtils.parseDurationUnit(argument)).isEqualTo(expected);
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"2 weeks", "3 months", "1 year", "fortnight", "days"})
+  void givenUnsupportedUnitOrFormat_whenParseDurationUnit_thenThrows(String argument) {
+    assertThatThrownBy(() -> TimeUtils.parseDurationUnit(argument))
+        .isInstanceOfAny(IllegalArgumentException.class, NumberFormatException.class);
+  }
+}
diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/partitionTest.sqrl b/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/partitionTest.sqrl
index 981e56553..be55bed47 100644
--- a/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/partitionTest.sqrl
+++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/partitionTest.sqrl
@@ -5,3 +5,6 @@ OrderUpdates := SELECT * FROM _OrdersStream;
 
 /*+ partition_key(id) */
 Orders := DISTINCT _OrdersStream ON id ORDER BY `time` DESC;
+
+/*+ partition_key(time), ttl(48 hours) */
+OrderMetrics := SELECT * FROM _OrdersStream;
diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/partitionTest.txt b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/partitionTest.txt
index a1d673d97..ade4ef2a9 100644
--- a/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/partitionTest.txt
+++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/partitionTest.txt
@@ -1,4 +1,27 @@
 >>>pipeline_explain.txt
+=== OrderMetrics
+ID:          default_catalog.default_database.OrderMetrics
+Type:        stream
+Stage:       flink
+Primary key: id, time
+Timestamp:   time
+Row count:   ~1e8
+---
+Schema:
+ - id: BIGINT NOT NULL
+ - customerid: BIGINT NOT NULL
+ - time: TIMESTAMP_LTZ(3) *ROWTIME* NOT NULL
+ - entries: RecordType:peek_no_expand(BIGINT NOT NULL productid, BIGINT NOT NULL quantity, DOUBLE NOT NULL unit_price, DOUBLE discount) NOT NULL ARRAY NOT NULL
+Inputs:
+ - default_catalog.default_database._OrdersStream
+Annotations:
+ - stream-root: _OrdersStream
+Plan:
+LogicalProject(id=[$0], customerid=[$1], time=[$2], entries=[$3])
+  LogicalTableScan(table=[[default_catalog, default_database, _OrdersStream]])
+SQL:
+CREATE VIEW `OrderMetrics` AS  SELECT * FROM _OrdersStream;
+
 === OrderUpdates
 ID:          default_catalog.default_database.OrderUpdates
 Type:        stream
@@ -113,7 +136,27 @@ SELECT `id`, `customerid`, `time`, `entries`
 FROM (SELECT `id`, `customerid`, `time`, `entries`, ROW_NUMBER() OVER (PARTITION BY `id` ORDER BY `time` DESC NULLS LAST) AS `__sqrlinternal_rownum`
   FROM `default_catalog`.`default_database`.`_OrdersStream`) AS `t`
 WHERE `__sqrlinternal_rownum` = 1;
-CREATE TABLE `OrderUpdates_1` (
+CREATE VIEW `OrderMetrics`
+AS
+SELECT *
+FROM `_OrdersStream`;
+CREATE TABLE `OrderMetrics_1` (
+  `id` BIGINT NOT NULL,
+  `customerid` BIGINT NOT NULL,
+  `time` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL,
+  `entries` RAW('com.datasqrl.flinkrunner.stdlib.json.FlinkJsonType', 'AERjb20uZGF0YXNxcmwuZmxpbmtydW5uZXIuc3RkbGliLmpzb24uRmxpbmtKc29uVHlwZVNlcmlhbGl6ZXJTbmFwc2hvdAAAAAM='),
+  PRIMARY KEY (`id`, `time`) NOT ENFORCED
+)
+WITH (
+  'connector' = 'jdbc-sqrl',
+  'driver' = 'org.postgresql.Driver',
+  'password' = '${POSTGRES_PASSWORD}',
+  'sink.on-conflict.action' = 'IGNORE',
+  'table-name' = 'OrderMetrics_1',
+  'url' = 'jdbc:postgresql://${POSTGRES_AUTHORITY}',
+  'username' = '${POSTGRES_USERNAME}'
+);
+CREATE TABLE `OrderUpdates_2` (
   `id` BIGINT NOT NULL,
   `customerid` BIGINT NOT NULL,
   `time` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL,
@@ -125,11 +168,11 @@ WITH (
   'driver' = 'org.postgresql.Driver',
   'password' = '${POSTGRES_PASSWORD}',
   'sink.on-conflict.action' = 'IGNORE',
-  'table-name' = 'OrderUpdates_1',
+  'table-name' = 'OrderUpdates_2',
   'url' = 'jdbc:postgresql://${POSTGRES_AUTHORITY}',
   'username' = '${POSTGRES_USERNAME}'
 );
-CREATE TABLE `Orders_2` (
+CREATE TABLE `Orders_3` (
   `id` BIGINT NOT NULL,
   `customerid` BIGINT NOT NULL,
   `time` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL,
@@ -142,16 +185,20 @@ WITH (
   'password' = '${POSTGRES_PASSWORD}',
   'sink.on-conflict.action' = 'TIMESTAMP',
   'sink.on-conflict.timestamp-column' = 'time',
-  'table-name' = 'Orders_2',
+  'table-name' = 'Orders_3',
   'url' = 'jdbc:postgresql://${POSTGRES_AUTHORITY}',
   'username' = '${POSTGRES_USERNAME}'
 );
 EXECUTE STATEMENT SET BEGIN
-INSERT INTO `default_catalog`.`default_database`.`OrderUpdates_1`
+INSERT INTO `default_catalog`.`default_database`.`OrderMetrics_1`
+SELECT `id`, `customerid`, `time`, `to_jsonb`(`entries`) AS `entries`
+FROM `default_catalog`.`default_database`.`OrderMetrics`
+;
+INSERT INTO `default_catalog`.`default_database`.`OrderUpdates_2`
 SELECT `id`, `customerid`, `time`, `to_jsonb`(`entries`) AS `entries`
 FROM `default_catalog`.`default_database`.`OrderUpdates`
 ;
-INSERT INTO `default_catalog`.`default_database`.`Orders_2`
+INSERT INTO `default_catalog`.`default_database`.`Orders_3`
 SELECT `id`, `customerid`, `time`, `to_jsonb`(`entries`) AS `entries`
 FROM `default_catalog`.`default_database`.`_OrdersStream`
 ;
@@ -165,9 +212,46 @@ END
 {
   "statements" : [
     {
-      "name" : "OrderUpdates_1",
+      "name" : "OrderMetrics_1",
       "type" : "TABLE",
-      "sql" : "CREATE TABLE IF NOT EXISTS \"OrderUpdates_1\" (\"id\" BIGINT NOT NULL, \"customerid\" BIGINT NOT NULL, \"time\" TIMESTAMP WITH TIME ZONE NOT NULL, \"entries\" JSONB, PRIMARY KEY (\"id\",\"time\")) PARTITION BY RANGE (\"time\")",
+      "sql" : "CREATE TABLE IF NOT EXISTS \"OrderMetrics_1\" (\"id\" BIGINT NOT NULL, \"customerid\" BIGINT NOT NULL, \"time\" TIMESTAMP WITH TIME ZONE NOT NULL, \"entries\" JSONB, PRIMARY KEY (\"id\",\"time\")) PARTITION BY RANGE (\"time\")",
+      "fields" : [
+        {
+          "name" : "id",
+          "type" : "BIGINT",
+          "nullable" : false
+        },
+        {
+          "name" : "customerid",
+          "type" : "BIGINT",
+          "nullable" : false
+        },
+        {
+          "name" : "time",
+          "type" : "TIMESTAMP WITH TIME ZONE",
+          "nullable" : false
+        },
+        {
+          "name" : "entries",
+          "type" : "JSONB",
+          "nullable" : true
+        }
+      ],
+      "primaryKey" : [
+        "id",
+        "time"
+      ],
+      "partitionKey" : [
+        "time"
+      ],
+      "partitionType" : "RANGE",
+      "numPartitions" : 1,
+      "ttl" : 172800.000000000
+    },
+    {
+      "name" : "OrderUpdates_2",
+      "type" : "TABLE",
+      "sql" : "CREATE TABLE IF NOT EXISTS \"OrderUpdates_2\" (\"id\" BIGINT NOT NULL, \"customerid\" BIGINT NOT NULL, \"time\" TIMESTAMP WITH TIME ZONE NOT NULL, \"entries\" JSONB, PRIMARY KEY (\"id\",\"time\")) PARTITION BY RANGE (\"time\")",
       "fields" : [
         {
           "name" : "id",
@@ -202,9 +286,9 @@ END
       "ttl" : 2592000.000000000
     },
     {
-      "name" : "Orders_2",
+      "name" : "Orders_3",
       "type" : "TABLE",
-      "sql" : "CREATE TABLE IF NOT EXISTS \"Orders_2\" (\"id\" BIGINT NOT NULL, \"customerid\" BIGINT NOT NULL, \"time\" TIMESTAMP WITH TIME ZONE NOT NULL, \"entries\" JSONB, PRIMARY KEY (\"id\")) PARTITION BY HASH (\"id\");\n\nCREATE TABLE IF NOT EXISTS \"Orders_2_all\" PARTITION OF \"Orders_2\" FOR VALUES WITH (MODULUS 1, REMAINDER 0)",
+      "sql" : "CREATE TABLE IF NOT EXISTS \"Orders_3\" (\"id\" BIGINT NOT NULL, \"customerid\" BIGINT NOT NULL, \"time\" TIMESTAMP WITH TIME ZONE NOT NULL, \"entries\" JSONB, PRIMARY KEY (\"id\")) PARTITION BY HASH (\"id\");\n\nCREATE TABLE IF NOT EXISTS \"Orders_3_all\" PARTITION OF \"Orders_3\" FOR VALUES WITH (MODULUS 1, REMAINDER 0)",
       "fields" : [
         {
           "name" : "id",
@@ -237,10 +321,37 @@ END
       "numPartitions" : 1,
       "ttl" : 0.0
     },
+    {
+      "name" : "OrderMetrics",
+      "type" : "VIEW",
+      "sql" : "CREATE OR REPLACE VIEW \"OrderMetrics\"(\"id\", \"customerid\", \"time\", \"entries\") AS SELECT *\nFROM \"OrderMetrics_1\"",
+      "fields" : [
+        {
+          "name" : "id",
+          "type" : "BIGINT",
+          "nullable" : false
+        },
+        {
+          "name" : "customerid",
+          "type" : "BIGINT",
+          "nullable" : false
+        },
+        {
+          "name" : "time",
+          "type" : "TIMESTAMP WITH TIME ZONE",
+          "nullable" : false
+        },
+        {
+          "name" : "entries",
+          "type" : "JSONB",
+          "nullable" : true
+        }
+      ]
+    },
     {
       "name" : "OrderUpdates",
       "type" : "VIEW",
-      "sql" : "CREATE OR REPLACE VIEW \"OrderUpdates\"(\"id\", \"customerid\", \"time\", \"entries\") AS SELECT *\nFROM \"OrderUpdates_1\"",
+      "sql" : "CREATE OR REPLACE VIEW \"OrderUpdates\"(\"id\", \"customerid\", \"time\", \"entries\") AS SELECT *\nFROM \"OrderUpdates_2\"",
       "fields" : [
         {
           "name" : "id",
@@ -267,7 +378,7 @@ END
     {
       "name" : "Orders",
       "type" : "VIEW",
-      "sql" : "CREATE OR REPLACE VIEW \"Orders\"(\"id\", \"customerid\", \"time\", \"entries\") AS SELECT *\nFROM \"Orders_2\"",
+      "sql" : "CREATE OR REPLACE VIEW \"Orders\"(\"id\", \"customerid\", \"time\", \"entries\") AS SELECT *\nFROM \"Orders_3\"",
       "fields" : [
         {
           "name" : "id",
@@ -296,7 +407,7 @@ END
     {
       "name" : "partman",
       "type" : "EXTENSION",
-      "sql" : "CREATE SCHEMA IF NOT EXISTS partman;\nCREATE EXTENSION IF NOT EXISTS pg_partman SCHEMA partman;\n\nSELECT partman.create_parent(\n    p_parent_table => 'public.OrderUpdates_1',\n    p_control => 'time',\n    p_interval => '1 week',\n    p_premake => 4\n);\n\nUPDATE partman.part_config\n   SET retention = '30 days',\n       retention_keep_table = false\n WHERE parent_table = 'public.OrderUpdates_1';"
+      "sql" : "CREATE SCHEMA IF NOT EXISTS partman;\nCREATE EXTENSION IF NOT EXISTS pg_partman SCHEMA partman;\n\nSELECT partman.create_parent(\n    p_parent_table => 'public.OrderMetrics_1',\n    p_control => 'time',\n    p_interval => '1 hour',\n    p_premake => 4\n);\n\nUPDATE partman.part_config\n   SET retention = '2 days',\n       retention_keep_table = false\n WHERE parent_table = 'public.OrderMetrics_1';\n\nSELECT partman.create_parent(\n    p_parent_table => 'public.OrderUpdates_2',\n    p_control => 'time',\n    p_interval => '1 day',\n    p_premake => 4\n);\n\nUPDATE partman.part_config\n   SET retention = '30 days',\n       retention_keep_table = false\n WHERE parent_table = 'public.OrderUpdates_2';"
     }
   ]
 }
@@ -305,6 +416,31 @@ END
   "models" : {
     "v1" : {
       "queries" : [
+        {
+          "type" : "args",
+          "parentType" : "Query",
+          "fieldName" : "OrderMetrics",
+          "exec" : {
+            "arguments" : [
+              {
+                "type" : "variable",
+                "path" : "limit"
+              },
+              {
+                "type" : "variable",
+                "path" : "offset"
+              }
+            ],
+            "query" : {
+              "type" : "SqlQuery",
+              "sql" : "SELECT *\nFROM \"OrderMetrics_1\"",
+              "parameters" : [ ],
+              "pagination" : "LIMIT_AND_OFFSET",
+              "cacheDurationMs" : 0,
+              "database" : "POSTGRES"
+            }
+          }
+        },
         {
           "type" : "args",
           "parentType" : "Query",
@@ -322,7 +458,7 @@ END
             ],
             "query" : {
               "type" : "SqlQuery",
-              "sql" : "SELECT *\nFROM \"OrderUpdates_1\"",
+              "sql" : "SELECT *\nFROM \"OrderUpdates_2\"",
               "parameters" : [ ],
               "pagination" : "LIMIT_AND_OFFSET",
               "cacheDurationMs" : 0,
@@ -347,7 +483,7 @@ END
             ],
             "query" : {
               "type" : "SqlQuery",
-              "sql" : "SELECT *\nFROM \"Orders_2\"",
+              "sql" : "SELECT *\nFROM \"Orders_3\"",
               "parameters" : [ ],
               "pagination" : "LIMIT_AND_OFFSET",
               "cacheDurationMs" : 0,
@@ -359,6 +495,32 @@ END
       "mutations" : [ ],
       "subscriptions" : [ ],
       "operations" : [
+        {
+          "function" : {
+            "name" : "GetOrderMetrics",
+            "parameters" : {
+              "type" : "object",
+              "properties" : {
+                "offset" : {
+                  "type" : "integer"
+                },
+                "limit" : {
+                  "type" : "integer"
+                }
+              },
+              "required" : [ ]
+            }
+          },
+          "format" : "JSON",
+          "apiQuery" : {
+            "query" : "query OrderMetrics($limit: Int = 10, $offset: Int = 0) {\nOrderMetrics(limit: $limit, offset: $offset) {\nid\ncustomerid\ntime\nentries {\nproductid\nquantity\nunit_price\ndiscount\n}\n}\n\n}",
+            "queryName" : "OrderMetrics",
+            "operationType" : "QUERY"
+          },
+          "mcpMethod" : "TOOL",
+          "restMethod" : "GET",
+          "uriTemplate" : "queries/OrderMetrics{?offset,limit}"
+        },
         {
           "function" : {
             "name" : "GetOrderUpdates",
@@ -414,7 +576,7 @@ END
       ],
       "schema" : {
         "type" : "string",
-        "schema" : "\"An RFC-3339 compliant Full Date Scalar\"\nscalar Date\n\n\"A DateTime scalar that handles both full RFC3339 and shorter timestamp formats\"\nscalar DateTime\n\n\"A JSON scalar\"\nscalar JSON\n\n\"24-hour clock time value string in the format `hh:mm:ss` or `hh:mm:ss.sss`.\"\nscalar LocalTime\n\n\"A 64-bit signed integer\"\nscalar Long\n\ntype OrderUpdates {\n  id: Long!\n  customerid: Long!\n  time: DateTime!\n  entries: [OrderUpdates_entriesOutput]!\n}\n\ntype OrderUpdates_entriesOutput {\n  productid: Long!\n  quantity: Long!\n  unit_price: Float!\n  discount: Float\n}\n\ntype Orders {\n  id: Long!\n  customerid: Long!\n  time: DateTime!\n  entries: [Orders_entriesOutput]!\n}\n\ntype Orders_entriesOutput {\n  productid: Long!\n  quantity: Long!\n  unit_price: Float!\n  discount: Float\n}\n\ntype Query {\n  OrderUpdates(limit: Int = 10, offset: Int = 0): [OrderUpdates!]\n  Orders(limit: Int = 10, offset: Int = 0): [Orders!]\n}\n\nenum _McpMethodType {\n  NONE\n  TOOL\n  RESOURCE\n}\n\nenum _RestMethodType {\n  NONE\n  GET\n  POST\n}\n\ndirective @api(mcp: _McpMethodType, rest: _RestMethodType, uri: String) on QUERY | MUTATION | FIELD_DEFINITION\n"
+        "schema" : "\"An RFC-3339 compliant Full Date Scalar\"\nscalar Date\n\n\"A DateTime scalar that handles both full RFC3339 and shorter timestamp formats\"\nscalar DateTime\n\n\"A JSON scalar\"\nscalar JSON\n\n\"24-hour clock time value string in the format `hh:mm:ss` or `hh:mm:ss.sss`.\"\nscalar LocalTime\n\n\"A 64-bit signed integer\"\nscalar Long\n\ntype OrderMetrics {\n  id: Long!\n  customerid: Long!\n  time: DateTime!\n  entries: [OrderMetrics_entriesOutput]!\n}\n\ntype OrderMetrics_entriesOutput {\n  productid: Long!\n  quantity: Long!\n  unit_price: Float!\n  discount: Float\n}\n\ntype OrderUpdates {\n  id: Long!\n  customerid: Long!\n  time: DateTime!\n  entries: [OrderUpdates_entriesOutput]!\n}\n\ntype OrderUpdates_entriesOutput {\n  productid: Long!\n  quantity: Long!\n  unit_price: Float!\n  discount: Float\n}\n\ntype Orders {\n  id: Long!\n  customerid: Long!\n  time: DateTime!\n  entries: [Orders_entriesOutput]!\n}\n\ntype Orders_entriesOutput {\n  productid: Long!\n  quantity: Long!\n  unit_price: Float!\n  discount: Float\n}\n\ntype Query {\n  OrderMetrics(limit: Int = 10, offset: Int = 0): [OrderMetrics!]\n  OrderUpdates(limit: Int = 10, offset: Int = 0): [OrderUpdates!]\n  Orders(limit: Int = 10, offset: Int = 0): [Orders!]\n}\n\nenum _McpMethodType {\n  NONE\n  TOOL\n  RESOURCE\n}\n\nenum _RestMethodType {\n  NONE\n  GET\n  POST\n}\n\ndirective @api(mcp: _McpMethodType, rest: _RestMethodType, uri: String) on QUERY | MUTATION | FIELD_DEFINITION\n"
       }
     }
   }