From acc6465f39e74e6d851474b093c5b3d449a115bf Mon Sep 17 00:00:00 2001 From: Wellington Mafra Date: Mon, 27 Jul 2026 15:31:25 -0300 Subject: [PATCH 1/8] feat: Support explicit partition interval as optional second argument in ttl hint Signed-off-by: Wellington Mafra --- documentation/docs/sqrl-language.md | 2 +- .../AbstractJdbcStatementFactory.java | 11 +- .../relational/CreateTableJdbcStatement.java | 7 +- .../extensions/PgPartmanExtension.java | 6 +- .../com/datasqrl/planner/hint/TtlHint.java | 54 ++++- .../PostgresCreateTableDdlFactoryTest.java | 3 +- .../extensions/PgPartmanExtensionTest.java | 44 +++- .../datasqrl/planner/hint/TtlHintTest.java | 101 +++++++++ .../resources/dagplanner/partitionTest.sqrl | 3 + .../datasqrl/DAGPlannerTest/partitionTest.txt | 195 ++++++++++++++++-- 10 files changed, 394 insertions(+), 32 deletions(-) create mode 100644 sqrl-planner/src/test/java/com/datasqrl/planner/hint/TtlHintTest.java diff --git a/documentation/docs/sqrl-language.md b/documentation/docs/sqrl-language.md index 219cbd0787..3a21b90a09 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 [, partition interval])` | 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`. An optional second argument sets an explicit partition interval (e.g. `ttl(14 days, 1 day)`) for range-partitioned tables; when omitted, the interval is derived from the duration. 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 cdcd374abd..80f507b463 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 @@ -160,13 +160,9 @@ 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); + String partitionInterval = ttlHint.flatMap(TtlHint::getPartitionInterval).orElse(null); return new CreateTableJdbcStatement( tableName, @@ -180,6 +176,7 @@ public JdbcStatement createTable(JdbcEngineCreateTable createTable) { partitionType, partitionType == PartitionType.NONE ? 0 : 1, ttl, + partitionInterval, createTable, createTableDdlFactory); } 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 d216243ea5..ef60e2e668 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 @@ -60,6 +60,9 @@ public class CreateTableJdbcStatement implements JdbcStatement { /** The time-to-live of records in this table - ZERO to disable */ Duration ttl; + /** Explicit partition interval for range-partitioned tables - null to derive from ttl */ + String partitionInterval; + /** The engine table - nullable and not set when deserialized */ @JsonIgnore JdbcEngineCreateTable engineTable; @@ -75,7 +78,8 @@ public CreateTableJdbcStatement( @JsonProperty("partitionKey") List 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; @@ -84,6 +88,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/function/translation/postgres/extensions/PgPartmanExtension.java b/sqrl-planner/src/main/java/com/datasqrl/function/translation/postgres/extensions/PgPartmanExtension.java index 64e1162a07..fffe60b572 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 @@ -60,6 +60,10 @@ 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 = + createTable.getPartitionInterval() != null + ? createTable.getPartitionInterval() + : deriveInterval(ttl); // p_type was removed in pg_partman 5.x; the default (range) is what we need sb.append( @@ -72,7 +76,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( """ 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 b8fa3cf4ee..e2bea8d16f 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 @@ -22,29 +22,58 @@ import com.google.auto.service.AutoService; import java.time.Duration; import java.util.Optional; +import java.util.regex.Pattern; import org.apache.flink.util.TimeUtils; public class TtlHint extends PlannerHint { public static final String HINT_NAME = "ttl"; + /** Subset of postgres interval syntax accepted as an explicit partition interval */ + private static final Pattern PARTITION_INTERVAL_PATTERN = + Pattern.compile( + "^\\d+\\s*(second|minute|hour|day|week|month|year)s?$", Pattern.CASE_INSENSITIVE); + private final Duration ttl; + private final String partitionInterval; - protected TtlHint(ParsedObject source, Duration ttlDuration) { + protected TtlHint(ParsedObject source, Duration ttlDuration, String partitionInterval) { super(source, Type.DAG); this.ttl = ttlDuration; + this.partitionInterval = partitionInterval; } public Optional getTtl() { return Optional.ofNullable(ttl); } + public Optional getPartitionInterval() { + return Optional.ofNullable(partitionInterval); + } + @AutoService(Factory.class) public static class TtlHintFactory implements Factory { @Override public PlannerHint create(ParsedObject 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() > 2 || arguments.get(0) == null) { + throw new StatementParserException( + ErrorLabel.GENERIC, + source.getFileLocation(), + "%s hint supports a duration argument and an optional partition interval argument" + + " (e.g. `14 days, 1 day`).", + source.get().name()); + } + var ttl = parseDurationArgument(source, arguments.get(0)); + String partitionInterval = null; + if (arguments.size() == 2) { + partitionInterval = parsePartitionInterval(source, arguments.get(1)); + } + return new TtlHint(source, ttl, partitionInterval); } @Override @@ -65,8 +94,12 @@ public static Duration parseDuration(ParsedObject source) { "%s hint only supports one duration argument (e.g. `2 days`).", source.get().name()); } + return parseDurationArgument(source, arguments.get(0)); + } + + private static Duration parseDurationArgument(ParsedObject source, String argument) { try { - return TimeUtils.parseDuration(arguments.get(0)); + return TimeUtils.parseDuration(argument); } catch (Exception e) { throw new StatementParserException( ErrorLabel.GENERIC, @@ -74,7 +107,20 @@ public static Duration parseDuration(ParsedObject source) { "%s hint does not have a valid duration argument: %s. Expected `2 days` or `10 s`. " + e.getMessage(), source.get().name(), - arguments.get(0)); + argument); + } + } + + private static String parsePartitionInterval(ParsedObject source, String argument) { + if (argument == null || !PARTITION_INTERVAL_PATTERN.matcher(argument).matches()) { + throw new StatementParserException( + ErrorLabel.GENERIC, + source.getFileLocation(), + "%s hint does not have a valid partition interval argument: %s. Expected an interval" + + " like `1 day` or `1 month`.", + source.get().name(), + argument); } + return argument; } } 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 f7b7377d51..7976a814ec 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 2f8b5c20cb..35a21547a7 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 partitionKey, Duration ttl) { + return table(name, partitionType, partitionKey, ttl, null); + } + + private static CreateTableJdbcStatement table( + String name, + PartitionType partitionType, + List 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 @@ -82,6 +99,31 @@ void givenMixedTables_whenGetDdl_thenOnlyRangeTtlTablesSortedByName() { assertThat(sql).doesNotContain("p_type"); } + @Test + void givenExplicitPartitionInterval_whenGetDdl_thenIntervalOverridesDerived() { + assertThat( + extension.getDdl( + List.of( + table( + "Metrics", + PartitionType.RANGE, + List.of("time"), + Duration.ofDays(14), + "1 day")))) + .contains("p_interval => '1 day'") + .contains("retention = '14 days'"); + } + + @Test + void givenNoPartitionInterval_whenGetDdl_thenIntervalDerivedFromTtl() { + assertThat( + extension.getDdl( + List.of( + table("Metrics", PartitionType.RANGE, List.of("time"), Duration.ofDays(14))))) + .contains("p_interval => '1 week'") + .contains("retention = '14 days'"); + } + @Test void givenTtl_whenDeriveInterval_thenBucketedByDuration() { assertThat(PgPartmanExtension.deriveInterval(Duration.ofHours(6))).isEqualTo("1 day"); 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 0000000000..030052d62c --- /dev/null +++ b/sqrl-planner/src/test/java/com/datasqrl/planner/hint/TtlHintTest.java @@ -0,0 +1,101 @@ +/* + * 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.util.List; +import org.junit.jupiter.api.Test; + +class TtlHintTest { + + private final TtlHint.TtlHintFactory ttlFactory = new TtlHint.TtlHintFactory(); + private final CacheHint.CacheHintFactory cacheFactory = new CacheHint.CacheHintFactory(); + + private static ParsedObject hint(String name, String... args) { + return new ParsedObject<>(new SqrlHint(name, List.of(args)), FileLocation.START); + } + + @Test + void givenSingleDurationArg_whenCreate_thenTtlSetAndNoPartitionInterval() { + var hint = (TtlHint) ttlFactory.create(hint("ttl", "14 days")); + + assertThat(hint.getTtl()).contains(Duration.ofDays(14)); + assertThat(hint.getPartitionInterval()).isEmpty(); + } + + @Test + void givenNoArgs_whenCreate_thenEmptyTtlAndPartitionInterval() { + var hint = (TtlHint) ttlFactory.create(hint("ttl")); + + assertThat(hint.getTtl()).isEmpty(); + assertThat(hint.getPartitionInterval()).isEmpty(); + } + + @Test + void givenDurationAndPartitionIntervalArgs_whenCreate_thenBothSet() { + var hint = (TtlHint) ttlFactory.create(hint("ttl", "14 days", "1 day")); + + assertThat(hint.getTtl()).contains(Duration.ofDays(14)); + assertThat(hint.getPartitionInterval()).contains("1 day"); + } + + @Test + void givenPluralPartitionInterval_whenCreate_thenAccepted() { + var hint = (TtlHint) ttlFactory.create(hint("ttl", "90 days", "2 weeks")); + + assertThat(hint.getPartitionInterval()).contains("2 weeks"); + } + + @Test + void givenInvalidPartitionInterval_whenCreate_thenThrows() { + assertThatThrownBy(() -> ttlFactory.create(hint("ttl", "14 days", "daily"))) + .isInstanceOf(StatementParserException.class) + .hasMessageContaining("partition interval"); + } + + @Test + void givenThreeArgs_whenCreate_thenThrows() { + assertThatThrownBy(() -> ttlFactory.create(hint("ttl", "14 days", "1 day", "extra"))) + .isInstanceOf(StatementParserException.class); + } + + @Test + void givenInvalidDuration_whenCreate_thenThrows() { + assertThatThrownBy(() -> ttlFactory.create(hint("ttl", "fortnight"))) + .isInstanceOf(StatementParserException.class) + .hasMessageContaining("duration"); + } + + @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); + } +} 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 981e565538..4dae3a4b3d 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(14 days, 1 day) */ +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 a1d673d979..31e2f06597 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,47 @@ 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" : 1209600.000000000, + "partitionInterval" : "1 day" + }, + { + "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 +287,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 +322,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 +379,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 +408,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 day',\n p_premake => 4\n);\n\nUPDATE partman.part_config\n SET retention = '14 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 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_2';" } ] } @@ -305,6 +417,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 +459,7 @@ END ], "query" : { "type" : "SqlQuery", - "sql" : "SELECT *\nFROM \"OrderUpdates_1\"", + "sql" : "SELECT *\nFROM \"OrderUpdates_2\"", "parameters" : [ ], "pagination" : "LIMIT_AND_OFFSET", "cacheDurationMs" : 0, @@ -347,7 +484,7 @@ END ], "query" : { "type" : "SqlQuery", - "sql" : "SELECT *\nFROM \"Orders_2\"", + "sql" : "SELECT *\nFROM \"Orders_3\"", "parameters" : [ ], "pagination" : "LIMIT_AND_OFFSET", "cacheDurationMs" : 0, @@ -359,6 +496,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 +577,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" } } } From 43f943dcb224783b1128e04c1ae09aa5e4c955c6 Mon Sep 17 00:00:00 2001 From: Wellington Mafra Date: Mon, 27 Jul 2026 15:57:38 -0300 Subject: [PATCH 2/8] chore: Add Wellington Mafra to CONTRIBUTORS.md Signed-off-by: Wellington Mafra --- CONTRIBUTORS.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index 318d5e4dae..9a5e0bb628 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 | From 3a36c731e9c29b9440cf795488e5b3eb92fb482e Mon Sep 17 00:00:00 2001 From: Wellington Mafra Date: Mon, 27 Jul 2026 16:06:17 -0300 Subject: [PATCH 3/8] test: Cover ttl hint null branches and factory ttl extraction Signed-off-by: Wellington Mafra --- .../AbstractJdbcStatementFactoryTest.java | 92 +++++++++++++++++++ .../datasqrl/planner/hint/TtlHintTest.java | 40 ++++++++ 2 files changed, 132 insertions(+) create mode 100644 sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactoryTest.java diff --git a/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactoryTest.java b/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactoryTest.java new file mode 100644 index 0000000000..0d76a3af8e --- /dev/null +++ b/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactoryTest.java @@ -0,0 +1,92 @@ +/* + * 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.mockito.Mockito.when; + +import com.datasqrl.error.ErrorCollector; +import com.datasqrl.error.ErrorLocation.FileLocation; +import com.datasqrl.planner.analyzer.TableAnalysis; +import com.datasqrl.planner.hint.PlannerHints; +import com.datasqrl.planner.parser.ParsedObject; +import com.datasqrl.planner.parser.SqrlComments; +import com.datasqrl.planner.parser.SqrlHint; +import com.datasqrl.planner.tables.FlinkTableBuilder; +import com.datasqrl.planner.util.Documented; +import java.time.Duration; +import java.util.List; +import java.util.Optional; +import org.apache.calcite.rel.type.RelDataType; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class AbstractJdbcStatementFactoryTest { + + @Mock private TableAnalysis tableAnalysis; + + @Mock private RelDataType datatype; + + @Mock private FlinkTableBuilder tableBuilder; + + private final PostgresStatementFactory factory = new PostgresStatementFactory(); + + private CreateTableJdbcStatement createTable(PlannerHints hints) { + when(tableAnalysis.getHints()).thenReturn(hints); + when(tableAnalysis.getDocumentation()).thenReturn(Documented.EMPTY); + when(datatype.getFieldList()).thenReturn(List.of()); + when(tableBuilder.getPrimaryKey()).thenReturn(Optional.empty()); + + var createTable = new JdbcEngineCreateTable("my_table", tableBuilder, datatype, tableAnalysis); + return (CreateTableJdbcStatement) factory.createTable(createTable); + } + + private static PlannerHints hints(SqrlHint... sqrlHints) { + var parsedHints = + List.of(sqrlHints).stream() + .map(hint -> new ParsedObject<>(hint, FileLocation.START)) + .toList(); + var comments = new SqrlComments(List.of(), parsedHints); + return PlannerHints.from(comments, Optional.empty(), ErrorCollector.root()); + } + + @Test + void givenTtlHintWithPartitionInterval_whenCreateTable_thenBothPopulated() { + var stmt = createTable(hints(new SqrlHint("ttl", List.of("14 days", "1 day")))); + + assertThat(stmt.getTtl()).isEqualTo(Duration.ofDays(14)); + assertThat(stmt.getPartitionInterval()).isEqualTo("1 day"); + } + + @Test + void givenTtlHintWithoutPartitionInterval_whenCreateTable_thenIntervalNull() { + var stmt = createTable(hints(new SqrlHint("ttl", List.of("30 days")))); + + assertThat(stmt.getTtl()).isEqualTo(Duration.ofDays(30)); + assertThat(stmt.getPartitionInterval()).isNull(); + } + + @Test + void givenNoTtlHint_whenCreateTable_thenZeroTtlAndNullInterval() { + var stmt = createTable(PlannerHints.EMPTY); + + assertThat(stmt.getTtl()).isEqualTo(Duration.ZERO); + assertThat(stmt.getPartitionInterval()).isNull(); + } +} 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 index 030052d62c..c770f3ab69 100644 --- a/sqrl-planner/src/test/java/com/datasqrl/planner/hint/TtlHintTest.java +++ b/sqrl-planner/src/test/java/com/datasqrl/planner/hint/TtlHintTest.java @@ -23,6 +23,7 @@ import com.datasqrl.planner.parser.SqrlHint; import com.datasqrl.planner.parser.StatementParserException; import java.time.Duration; +import java.util.Arrays; import java.util.List; import org.junit.jupiter.api.Test; @@ -98,4 +99,43 @@ void givenTwoArgs_whenCreateCacheHint_thenThrows() { assertThatThrownBy(() -> cacheFactory.create(hint("cache", "5 min", "1 day"))) .isInstanceOf(StatementParserException.class); } + + @Test + void givenNullOptions_whenCreate_thenEmptyTtlAndPartitionInterval() { + var hint = (TtlHint) ttlFactory.create(nullArgsHint("ttl", (List) null)); + + assertThat(hint.getTtl()).isEmpty(); + assertThat(hint.getPartitionInterval()).isEmpty(); + } + + @Test + void givenNullOptions_whenCreateCacheHint_thenZeroDuration() { + var hint = (CacheHint) cacheFactory.create(nullArgsHint("cache", (List) 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); + } + + @Test + void givenNullPartitionInterval_whenCreate_thenThrows() { + assertThatThrownBy(() -> ttlFactory.create(nullArgsHint("ttl", Arrays.asList("14 days", null)))) + .isInstanceOf(StatementParserException.class) + .hasMessageContaining("partition interval"); + } + + private static ParsedObject nullArgsHint(String name, List args) { + return new ParsedObject<>(new SqrlHint(name, args), FileLocation.START); + } } From 23afed9cb4d974ccd7faa94c6478f15194a30173 Mon Sep 17 00:00:00 2001 From: Wellington Mafra Date: Tue, 28 Jul 2026 17:59:30 -0300 Subject: [PATCH 4/8] refactor: Derive partition width from ttl and partition-divisor setting --- .../docs/configuration-engine/postgres.md | 15 +++ documentation/docs/sqrl-language.md | 2 +- .../AbstractJdbcStatementFactory.java | 13 +- .../relational/CreateTableJdbcStatement.java | 2 +- .../relational/PostgresJdbcEngine.java | 2 +- .../relational/PostgresStatementFactory.java | 68 ++++++++++ .../extensions/PgPartmanExtension.java | 20 +-- .../com/datasqrl/planner/hint/TtlHint.java | 89 ++++++++----- .../AbstractJdbcStatementFactoryTest.java | 11 +- .../PostgresStatementFactoryTest.java | 122 ++++++++++++++++++ .../extensions/PgPartmanExtensionTest.java | 30 +---- .../datasqrl/planner/hint/TtlHintTest.java | 84 ++++++------ .../resources/dagplanner/partitionTest.sqrl | 2 +- .../datasqrl/DAGPlannerTest/partitionTest.txt | 9 +- 14 files changed, 336 insertions(+), 133 deletions(-) create mode 100644 sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/PostgresStatementFactoryTest.java diff --git a/documentation/docs/configuration-engine/postgres.md b/documentation/docs/configuration-engine/postgres.md index bc72e3e07a..4419fe4c53 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-divisor` | **integer** | `100` | Controls the number of partitions for range-partitioned tables with a TTL (see [Partitioning](#partitioning)). | + ## Basic Configuration ```json { "engines": { "postgres": { + "partition-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-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 3a21b90a09..cc4cae0735 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 [, partition interval])` | 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`. An optional second argument sets an explicit partition interval (e.g. `ttl(14 days, 1 day)`) for range-partitioned tables; when omitted, the interval is derived from the duration. 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 80f507b463..8fa86f1dab 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; @@ -162,7 +163,8 @@ public JdbcStatement createTable(JdbcEngineCreateTable createTable) { PartitionType partitionType = getPartitionType(createTable, partitionKeys); var ttlHint = createTable.tableAnalysis().getHints().getHint(TtlHint.class); Duration ttl = ttlHint.flatMap(TtlHint::getTtl).orElse(Duration.ZERO); - String partitionInterval = ttlHint.flatMap(TtlHint::getPartitionInterval).orElse(null); + ChronoUnit ttlUnit = ttlHint.flatMap(TtlHint::getTtlUnit).orElse(null); + String partitionInterval = derivePartitionInterval(partitionType, ttl, ttlUnit); return new CreateTableJdbcStatement( tableName, @@ -186,6 +188,15 @@ protected PartitionType getPartitionType( return partitionKey.isEmpty() ? PartitionType.NONE : PartitionType.LIST; } + /** + * The width of the partitions for range-partitioned tables with a TTL - null when the dialect + * does not support time-based partitioning. + */ + protected String derivePartitionInterval( + PartitionType partitionType, Duration ttl, ChronoUnit ttlUnit) { + return null; + } + protected List getColumns( List 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 ef60e2e668..ccf0b541b3 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 @@ -60,7 +60,7 @@ public class CreateTableJdbcStatement implements JdbcStatement { /** The time-to-live of records in this table - ZERO to disable */ Duration ttl; - /** Explicit partition interval for range-partitioned tables - null to derive from 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 */ 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 1775e469e3..c9abacb9aa 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/PostgresStatementFactory.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresStatementFactory.java index dc464c4a85..bba02928ad 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 @@ -23,6 +23,7 @@ import com.datasqrl.calcite.convert.PostgresSqlNodeToString; import com.datasqrl.calcite.dialect.ExtendedPostgresSqlDialect; import com.datasqrl.config.JdbcDialect; +import com.datasqrl.config.PackageJson.EngineConfig; import com.datasqrl.engine.database.relational.CreateTableJdbcStatement.PartitionType; import com.datasqrl.engine.database.relational.JdbcStatement.Type; import com.datasqrl.engine.database.relational.ddl.CreateIndexDDL; @@ -37,6 +38,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; @@ -50,12 +53,51 @@ public class PostgresStatementFactory extends AbstractJdbcStatementFactory implements JdbcStatementFactory { + public static final String PARTITION_DIVISOR_KEY = "partition-divisor"; + public static final int DEFAULT_PARTITION_DIVISOR = 100; + + /** + * Calendar-aligned partition widths pg_partman can use, in minutes, with their interval labels. + */ + private static final long[] PARTITION_MENU_MINUTES = { + 15, 30, 60, 120, 240, 360, 480, 720, 1440, 2880, 5760, 10080, 20160, 40320, 80640, 120960 + }; + + private static final String[] PARTITION_MENU_LABELS = { + "15 minutes", "30 minutes", "1 hour", "2 hours", "4 hours", "6 hours", "8 hours", "12 hours", + "1 day", "2 days", "4 days", "1 week", "2 weeks", "4 weeks", "8 weeks", "12 weeks" + }; + + private final int partitionDivisor; + public PostgresStatementFactory() { + this(DEFAULT_PARTITION_DIVISOR); + } + + public PostgresStatementFactory(EngineConfig engineConfig) { + this(parsePartitionDivisor(engineConfig)); + } + + public PostgresStatementFactory(int partitionDivisor) { super( new OperatorRuleTransformer(Dialect.POSTGRES), new PostgresRelToSqlNode(), new PostgresSqlNodeToString(), new PostgresCreateTableDdlFactory(true)); + checkArgument(partitionDivisor > 0, "%s must be a positive number", PARTITION_DIVISOR_KEY); + this.partitionDivisor = partitionDivisor; + } + + private static int parsePartitionDivisor(EngineConfig engineConfig) { + var setting = + engineConfig.getSetting( + PARTITION_DIVISOR_KEY, Optional.of(String.valueOf(DEFAULT_PARTITION_DIVISOR))); + try { + return Integer.parseInt(setting.trim()); + } catch (NumberFormatException e) { + throw new IllegalArgumentException( + "%s must be a positive number, but was: %s".formatted(PARTITION_DIVISOR_KEY, setting), e); + } } @Override @@ -100,6 +142,32 @@ protected PartitionType getPartitionType( return CalciteUtil.isTimestamp(colType.getType()) ? PartitionType.RANGE : PartitionType.HASH; } + @Override + protected String derivePartitionInterval( + PartitionType partitionType, Duration ttl, ChronoUnit ttlUnit) { + if (partitionType != PartitionType.RANGE || ttl == null || ttl.isZero() || ttlUnit == null) { + return null; + } + return derivePartitionInterval(ttl, ttlUnit, partitionDivisor); + } + + /** + * Picks the partition width for a range-partitioned table with a TTL: the TTL divided by the + * configured 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. + */ + static String derivePartitionInterval(Duration ttl, ChronoUnit ttlUnit, int partitionDivisor) { + var floorMinutes = ttlUnit.getDuration().toMinutes(); + var targetMinutes = Math.max(ttl.toMinutes() / (double) partitionDivisor, floorMinutes); + var interval = PARTITION_MENU_LABELS[0]; + for (var i = 0; i < PARTITION_MENU_MINUTES.length; i++) { + if (PARTITION_MENU_MINUTES[i] <= targetMinutes) { + interval = PARTITION_MENU_LABELS[i]; + } + } + return interval; + } + @Override public List applyTableExtensions(Collection tables) { var res = new ArrayList(); 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 fffe60b572..ee97854dbb 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.PartitionType; 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; @@ -61,9 +62,10 @@ private void appendTableDdl(StringBuilder sb, CreateTableJdbcStatement createTab var parentTable = "public." + createTable.getName(); var ttl = createTable.getTtl(); var interval = - createTable.getPartitionInterval() != null - ? createTable.getPartitionInterval() - : deriveInterval(ttl); + 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( @@ -96,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/TtlHint.java b/sqrl-planner/src/main/java/com/datasqrl/planner/hint/TtlHint.java index e2bea8d16f..c3d30a2b06 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 @@ -21,6 +21,9 @@ import com.datasqrl.planner.parser.StatementParserException; import com.google.auto.service.AutoService; import java.time.Duration; +import java.time.temporal.ChronoUnit; +import java.util.Locale; +import java.util.Map; import java.util.Optional; import java.util.regex.Pattern; import org.apache.flink.util.TimeUtils; @@ -29,26 +32,40 @@ public class TtlHint extends PlannerHint { public static final String HINT_NAME = "ttl"; - /** Subset of postgres interval syntax accepted as an explicit partition interval */ - private static final Pattern PARTITION_INTERVAL_PATTERN = - Pattern.compile( - "^\\d+\\s*(second|minute|hour|day|week|month|year)s?$", Pattern.CASE_INSENSITIVE); + private static final Pattern TTL_PATTERN = Pattern.compile("^(\\d+)\\s*([a-zA-Z]+)$"); + + /** Allowed TTL units: at least a minute, at most a week */ + private static final Map TTL_UNITS = + Map.ofEntries( + Map.entry("min", ChronoUnit.MINUTES), + Map.entry("minute", ChronoUnit.MINUTES), + Map.entry("minutes", ChronoUnit.MINUTES), + Map.entry("h", ChronoUnit.HOURS), + Map.entry("hour", ChronoUnit.HOURS), + Map.entry("hours", ChronoUnit.HOURS), + Map.entry("d", ChronoUnit.DAYS), + Map.entry("day", ChronoUnit.DAYS), + Map.entry("days", ChronoUnit.DAYS), + Map.entry("w", ChronoUnit.WEEKS), + Map.entry("week", ChronoUnit.WEEKS), + Map.entry("weeks", ChronoUnit.WEEKS)); private final Duration ttl; - private final String partitionInterval; + private final ChronoUnit ttlUnit; - protected TtlHint(ParsedObject source, Duration ttlDuration, String partitionInterval) { + protected TtlHint(ParsedObject source, Duration ttl, ChronoUnit ttlUnit) { super(source, Type.DAG); - this.ttl = ttlDuration; - this.partitionInterval = partitionInterval; + this.ttl = ttl; + this.ttlUnit = ttlUnit; } public Optional getTtl() { return Optional.ofNullable(ttl); } - public Optional getPartitionInterval() { - return Optional.ofNullable(partitionInterval); + /** The unit the TTL was declared with - determines the smallest allowed partition width */ + public Optional getTtlUnit() { + return Optional.ofNullable(ttlUnit); } @AutoService(Factory.class) @@ -60,20 +77,14 @@ public PlannerHint create(ParsedObject source) { if (arguments == null || arguments.isEmpty()) { return new TtlHint(source, null, null); } - if (arguments.size() > 2 || arguments.get(0) == null) { + if (arguments.size() != 1 || arguments.get(0) == null) { throw new StatementParserException( ErrorLabel.GENERIC, source.getFileLocation(), - "%s hint supports a duration argument and an optional partition interval argument" - + " (e.g. `14 days, 1 day`).", + "%s hint only supports one duration argument (e.g. `2 days`).", source.get().name()); } - var ttl = parseDurationArgument(source, arguments.get(0)); - String partitionInterval = null; - if (arguments.size() == 2) { - partitionInterval = parsePartitionInterval(source, arguments.get(1)); - } - return new TtlHint(source, ttl, partitionInterval); + return parseTtlArgument(source, arguments.get(0).trim()); } @Override @@ -82,6 +93,33 @@ public String getName() { } } + private static TtlHint parseTtlArgument(ParsedObject source, String argument) { + var matcher = TTL_PATTERN.matcher(argument); + ChronoUnit unit = null; + if (matcher.matches()) { + unit = TTL_UNITS.get(matcher.group(2).toLowerCase(Locale.ROOT)); + } + if (unit == null) { + throw new StatementParserException( + ErrorLabel.GENERIC, + source.getFileLocation(), + "%s hint does not have a valid duration argument: %s. Expected a positive number with a" + + " unit between minute and week, e.g. `30 min`, `36 hours`, `14 days`, or `2 weeks`.", + source.get().name(), + argument); + } + var value = Long.parseLong(matcher.group(1)); + var ttl = + switch (unit) { + case MINUTES -> Duration.ofMinutes(value); + case HOURS -> Duration.ofHours(value); + case DAYS -> Duration.ofDays(value); + case WEEKS -> Duration.ofDays(value * 7L); + default -> throw new IllegalStateException("Unexpected TTL unit: " + unit); + }; + return new TtlHint(source, ttl, unit); + } + public static Duration parseDuration(ParsedObject source) { var arguments = source.get().options(); if (arguments == null || arguments.isEmpty()) { @@ -110,17 +148,4 @@ private static Duration parseDurationArgument(ParsedObject source, Str argument); } } - - private static String parsePartitionInterval(ParsedObject source, String argument) { - if (argument == null || !PARTITION_INTERVAL_PATTERN.matcher(argument).matches()) { - throw new StatementParserException( - ErrorLabel.GENERIC, - source.getFileLocation(), - "%s hint does not have a valid partition interval argument: %s. Expected an interval" - + " like `1 day` or `1 month`.", - source.get().name(), - argument); - } - return argument; - } } diff --git a/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactoryTest.java b/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactoryTest.java index 0d76a3af8e..c9be05a089 100644 --- a/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactoryTest.java +++ b/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactoryTest.java @@ -67,18 +67,11 @@ private static PlannerHints hints(SqrlHint... sqrlHints) { } @Test - void givenTtlHintWithPartitionInterval_whenCreateTable_thenBothPopulated() { - var stmt = createTable(hints(new SqrlHint("ttl", List.of("14 days", "1 day")))); - - assertThat(stmt.getTtl()).isEqualTo(Duration.ofDays(14)); - assertThat(stmt.getPartitionInterval()).isEqualTo("1 day"); - } - - @Test - void givenTtlHintWithoutPartitionInterval_whenCreateTable_thenIntervalNull() { + void givenTtlHint_whenCreateTable_thenTtlPopulated() { var stmt = createTable(hints(new SqrlHint("ttl", List.of("30 days")))); assertThat(stmt.getTtl()).isEqualTo(Duration.ofDays(30)); + // no partition_key hint, so the table is not range-partitioned assertThat(stmt.getPartitionInterval()).isNull(); } 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 0000000000..eac0433283 --- /dev/null +++ b/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/PostgresStatementFactoryTest.java @@ -0,0 +1,122 @@ +/* + * 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.EmptyEngineConfig; +import com.datasqrl.config.PackageJson.EngineConfig; +import com.datasqrl.engine.database.relational.CreateTableJdbcStatement.PartitionType; +import java.time.Duration; +import java.time.temporal.ChronoUnit; +import java.util.Optional; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +class PostgresStatementFactoryTest { + + @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", + "2, WEEKS, 100, 1 week", + // 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_whenDeriveInterval_thenMenuWidth( + long amount, ChronoUnit unit, int divisor, String expected) { + var ttl = unit == ChronoUnit.WEEKS ? Duration.ofDays(amount * 7) : Duration.of(amount, unit); + + assertThat(PostgresStatementFactory.derivePartitionInterval(ttl, unit, divisor)) + .isEqualTo(expected); + } + + @Test + void givenNonRangeOrNoTtl_whenDeriveInterval_thenNull() { + var factory = new PostgresStatementFactory(); + + assertThat( + factory.derivePartitionInterval( + PartitionType.HASH, Duration.ofDays(14), ChronoUnit.DAYS)) + .isNull(); + assertThat(factory.derivePartitionInterval(PartitionType.RANGE, Duration.ZERO, ChronoUnit.DAYS)) + .isNull(); + assertThat(factory.derivePartitionInterval(PartitionType.RANGE, null, null)).isNull(); + assertThat(factory.derivePartitionInterval(PartitionType.RANGE, Duration.ofDays(14), null)) + .isNull(); + } + + @Test + void givenRangeAndTtl_whenDeriveInterval_thenWidthReturned() { + var factory = new PostgresStatementFactory(); + + assertThat( + factory.derivePartitionInterval( + PartitionType.RANGE, Duration.ofDays(14), ChronoUnit.DAYS)) + .isEqualTo("1 day"); + } + + @Test + void givenEmptyEngineConfig_whenCreate_thenDefaultDivisorUsed() { + var factory = new PostgresStatementFactory(new EmptyEngineConfig("postgres")); + + assertThat( + factory.derivePartitionInterval( + PartitionType.RANGE, Duration.ofDays(14), ChronoUnit.DAYS)) + .isEqualTo("1 day"); + } + + @Test + void givenConfiguredDivisor_whenCreate_thenDivisorApplied() { + var engineConfig = mock(EngineConfig.class); + when(engineConfig.getSetting( + PostgresStatementFactory.PARTITION_DIVISOR_KEY, Optional.of("100"))) + .thenReturn("4"); + var factory = new PostgresStatementFactory(engineConfig); + + assertThat( + factory.derivePartitionInterval( + PartitionType.RANGE, Duration.ofDays(84), ChronoUnit.DAYS)) + .isEqualTo("2 weeks"); + } + + @Test + void givenInvalidDivisor_whenCreate_thenThrows() { + var engineConfig = mock(EngineConfig.class); + when(engineConfig.getSetting( + PostgresStatementFactory.PARTITION_DIVISOR_KEY, Optional.of("100"))) + .thenReturn("not-a-number"); + + assertThatThrownBy(() -> new PostgresStatementFactory(engineConfig)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("partition-divisor"); + + assertThatThrownBy(() -> new PostgresStatementFactory(0)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("partition-divisor"); + } +} 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 35a21547a7..7f34b27598 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,7 +29,7 @@ class PgPartmanExtensionTest { private static CreateTableJdbcStatement table( String name, PartitionType partitionType, List partitionKey, Duration ttl) { - return table(name, partitionType, partitionKey, ttl, null); + return table(name, partitionType, partitionKey, ttl, "1 day"); } private static CreateTableJdbcStatement table( @@ -74,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'") @@ -100,7 +100,7 @@ void givenMixedTables_whenGetDdl_thenOnlyRangeTtlTablesSortedByName() { } @Test - void givenExplicitPartitionInterval_whenGetDdl_thenIntervalOverridesDerived() { + void givenPartitionInterval_whenGetDdl_thenUsedAsIs() { assertThat( extension.getDdl( List.of( @@ -109,31 +109,11 @@ void givenExplicitPartitionInterval_whenGetDdl_thenIntervalOverridesDerived() { PartitionType.RANGE, List.of("time"), Duration.ofDays(14), - "1 day")))) - .contains("p_interval => '1 day'") - .contains("retention = '14 days'"); - } - - @Test - void givenNoPartitionInterval_whenGetDdl_thenIntervalDerivedFromTtl() { - assertThat( - extension.getDdl( - List.of( - table("Metrics", PartitionType.RANGE, List.of("time"), Duration.ofDays(14))))) - .contains("p_interval => '1 week'") + "2 weeks")))) + .contains("p_interval => '2 weeks'") .contains("retention = '14 days'"); } - @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"); - } - @Test void givenTtl_whenFormatRetention_thenLargestWholeUnit() { assertThat(PgPartmanExtension.formatRetention(Duration.ofDays(30))).isEqualTo("30 days"); 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 index c770f3ab69..cc01ef27aa 100644 --- a/sqrl-planner/src/test/java/com/datasqrl/planner/hint/TtlHintTest.java +++ b/sqrl-planner/src/test/java/com/datasqrl/planner/hint/TtlHintTest.java @@ -23,9 +23,13 @@ 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 { @@ -37,54 +41,55 @@ private static ParsedObject hint(String name, String... args) { } @Test - void givenSingleDurationArg_whenCreate_thenTtlSetAndNoPartitionInterval() { + void givenSingleDurationArg_whenCreate_thenTtlAndUnitSet() { var hint = (TtlHint) ttlFactory.create(hint("ttl", "14 days")); assertThat(hint.getTtl()).contains(Duration.ofDays(14)); - assertThat(hint.getPartitionInterval()).isEmpty(); + assertThat(hint.getTtlUnit()).contains(ChronoUnit.DAYS); } @Test - void givenNoArgs_whenCreate_thenEmptyTtlAndPartitionInterval() { + void givenNoArgs_whenCreate_thenEmptyTtlAndUnit() { var hint = (TtlHint) ttlFactory.create(hint("ttl")); assertThat(hint.getTtl()).isEmpty(); - assertThat(hint.getPartitionInterval()).isEmpty(); - } - - @Test - void givenDurationAndPartitionIntervalArgs_whenCreate_thenBothSet() { - var hint = (TtlHint) ttlFactory.create(hint("ttl", "14 days", "1 day")); - - assertThat(hint.getTtl()).contains(Duration.ofDays(14)); - assertThat(hint.getPartitionInterval()).contains("1 day"); - } - - @Test - void givenPluralPartitionInterval_whenCreate_thenAccepted() { - var hint = (TtlHint) ttlFactory.create(hint("ttl", "90 days", "2 weeks")); - - assertThat(hint.getPartitionInterval()).contains("2 weeks"); - } - - @Test - void givenInvalidPartitionInterval_whenCreate_thenThrows() { - assertThatThrownBy(() -> ttlFactory.create(hint("ttl", "14 days", "daily"))) + 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", + "2 weeks, 20160, WEEKS", + "5 week, 50400, WEEKS", + "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", "3 months", "1 year", "fortnight", "14", "days"}) + void givenUnsupportedUnitOrFormat_whenCreate_thenThrows(String argument) { + assertThatThrownBy(() -> ttlFactory.create(hint("ttl", argument))) .isInstanceOf(StatementParserException.class) - .hasMessageContaining("partition interval"); - } - - @Test - void givenThreeArgs_whenCreate_thenThrows() { - assertThatThrownBy(() -> ttlFactory.create(hint("ttl", "14 days", "1 day", "extra"))) - .isInstanceOf(StatementParserException.class); + .hasMessageContaining("unit between minute and week"); } @Test - void givenInvalidDuration_whenCreate_thenThrows() { - assertThatThrownBy(() -> ttlFactory.create(hint("ttl", "fortnight"))) + void givenTwoArgs_whenCreate_thenThrows() { + assertThatThrownBy(() -> ttlFactory.create(hint("ttl", "14 days", "1 day"))) .isInstanceOf(StatementParserException.class) - .hasMessageContaining("duration"); + .hasMessageContaining("one duration argument"); } @Test @@ -101,11 +106,11 @@ void givenTwoArgs_whenCreateCacheHint_thenThrows() { } @Test - void givenNullOptions_whenCreate_thenEmptyTtlAndPartitionInterval() { + void givenNullOptions_whenCreate_thenEmptyTtlAndUnit() { var hint = (TtlHint) ttlFactory.create(nullArgsHint("ttl", (List) null)); assertThat(hint.getTtl()).isEmpty(); - assertThat(hint.getPartitionInterval()).isEmpty(); + assertThat(hint.getTtlUnit()).isEmpty(); } @Test @@ -128,13 +133,6 @@ void givenNullFirstArg_whenCreateCacheHint_thenThrows() { .isInstanceOf(StatementParserException.class); } - @Test - void givenNullPartitionInterval_whenCreate_thenThrows() { - assertThatThrownBy(() -> ttlFactory.create(nullArgsHint("ttl", Arrays.asList("14 days", null)))) - .isInstanceOf(StatementParserException.class) - .hasMessageContaining("partition interval"); - } - private static ParsedObject nullArgsHint(String name, List args) { return new ParsedObject<>(new SqrlHint(name, args), FileLocation.START); } 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 4dae3a4b3d..be55bed474 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 @@ -6,5 +6,5 @@ OrderUpdates := SELECT * FROM _OrdersStream; /*+ partition_key(id) */ Orders := DISTINCT _OrdersStream ON id ORDER BY `time` DESC; -/*+ partition_key(time), ttl(14 days, 1 day) */ +/*+ 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 31e2f06597..9d7ad8bdd3 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 @@ -246,8 +246,8 @@ END ], "partitionType" : "RANGE", "numPartitions" : 1, - "ttl" : 1209600.000000000, - "partitionInterval" : "1 day" + "ttl" : 172800.000000000, + "partitionInterval" : "1 hour" }, { "name" : "OrderUpdates_2", @@ -284,7 +284,8 @@ END ], "partitionType" : "RANGE", "numPartitions" : 1, - "ttl" : 2592000.000000000 + "ttl" : 2592000.000000000, + "partitionInterval" : "1 day" }, { "name" : "Orders_3", @@ -408,7 +409,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.OrderMetrics_1',\n p_control => 'time',\n p_interval => '1 day',\n p_premake => 4\n);\n\nUPDATE partman.part_config\n SET retention = '14 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 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_2';" + "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';" } ] } From 069d78ec37434ff9653635032f199095718fbc09 Mon Sep 17 00:00:00 2001 From: Wellington Mafra Date: Wed, 29 Jul 2026 13:38:56 -0300 Subject: [PATCH 5/8] refactor: Address pg_partman PR review feedback Signed-off-by: Wellington Mafra --- .../AbstractJdbcStatementFactory.java | 8 +- .../relational/PostgresPartitionInterval.java | 68 +++++++++++++++ .../relational/PostgresStatementFactory.java | 54 +----------- .../com/datasqrl/planner/hint/TtlHint.java | 47 +++------- .../java/com/datasqrl/util/TimeUtils.java | 58 +++++++++++++ .../src/main/resources/default-package.json | 3 + .../resources/jsonSchema/packageSchema.json | 4 + .../AbstractJdbcStatementFactoryTest.java | 85 ------------------- .../PostgresPartitionIntervalTest.java | 47 ++++++++++ .../PostgresStatementFactoryTest.java | 65 +++----------- .../datasqrl/planner/hint/TtlHintTest.java | 17 +++- .../java/com/datasqrl/util/TimeUtilsTest.java | 71 ++++++++++++++++ 12 files changed, 295 insertions(+), 232 deletions(-) create mode 100644 sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresPartitionInterval.java create mode 100644 sqrl-planner/src/main/java/com/datasqrl/util/TimeUtils.java delete mode 100644 sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactoryTest.java create mode 100644 sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/PostgresPartitionIntervalTest.java create mode 100644 sqrl-planner/src/test/java/com/datasqrl/util/TimeUtilsTest.java 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 5a9af4060e..73fce5bb4f 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 @@ -172,7 +172,7 @@ public JdbcStatement createTable(JdbcEngineCreateTable createTable) { 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); + String partitionInterval = derivePartitionInterval(partitionType, ttl, ttlUnit).orElse(null); return new CreateTableJdbcStatement( tableName, @@ -197,12 +197,12 @@ protected PartitionType getPartitionType( } /** - * The width of the partitions for range-partitioned tables with a TTL - null when the dialect + * The width of the partitions for range-partitioned tables with a TTL - empty when the dialect * does not support time-based partitioning. */ - protected String derivePartitionInterval( + protected Optional derivePartitionInterval( PartitionType partitionType, Duration ttl, ChronoUnit ttlUnit) { - return null; + return Optional.empty(); } protected List getColumns( 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 0000000000..a9e071a077 --- /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.Value; + +/** The pg_partman partition width for a range-partitioned table with a TTL. */ +@Value +public class PostgresPartitionInterval { + + /** Calendar-aligned partition widths pg_partman can use, keyed by width in minutes. */ + private static final Map PARTITION_MENU = + ImmutableMap.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(); + + String interval; + + /** + * 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 PostgresPartitionInterval of( + Duration ttl, ChronoUnit ttlUnit, int partitionDivisor) { + var floorMinutes = ttlUnit.getDuration().toMinutes(); + var targetMinutes = Math.max(ttl.toMinutes() / (double) partitionDivisor, floorMinutes); + var interval = PARTITION_MENU.values().iterator().next(); + for (var entry : PARTITION_MENU.entrySet()) { + if (entry.getKey() <= targetMinutes) { + interval = entry.getValue(); + } + } + return new PostgresPartitionInterval(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 b8573df7ad..fea87a3ed6 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 @@ -50,28 +50,11 @@ public class PostgresStatementFactory extends AbstractJdbcStatementFactory { public static final String PARTITION_DIVISOR_KEY = "partition-divisor"; - public static final int DEFAULT_PARTITION_DIVISOR = 100; - - /** - * Calendar-aligned partition widths pg_partman can use, in minutes, with their interval labels. - */ - private static final long[] PARTITION_MENU_MINUTES = { - 15, 30, 60, 120, 240, 360, 480, 720, 1440, 2880, 5760, 10080, 20160, 40320, 80640, 120960 - }; - - private static final String[] PARTITION_MENU_LABELS = { - "15 minutes", "30 minutes", "1 hour", "2 hours", "4 hours", "6 hours", "8 hours", "12 hours", - "1 day", "2 days", "4 days", "1 week", "2 weeks", "4 weeks", "8 weeks", "12 weeks" - }; private final int partitionDivisor; - public PostgresStatementFactory() { - this(DEFAULT_PARTITION_DIVISOR); - } - public PostgresStatementFactory(EngineConfig engineConfig) { - this(parsePartitionDivisor(engineConfig)); + this(Integer.parseInt(engineConfig.getSetting(PARTITION_DIVISOR_KEY))); } public PostgresStatementFactory(int partitionDivisor) { @@ -80,18 +63,6 @@ public PostgresStatementFactory(int partitionDivisor) { this.partitionDivisor = partitionDivisor; } - private static int parsePartitionDivisor(EngineConfig engineConfig) { - var setting = - engineConfig.getSetting( - PARTITION_DIVISOR_KEY, Optional.of(String.valueOf(DEFAULT_PARTITION_DIVISOR))); - try { - return Integer.parseInt(setting.trim()); - } catch (NumberFormatException e) { - throw new IllegalArgumentException( - "%s must be a positive number, but was: %s".formatted(PARTITION_DIVISOR_KEY, setting), e); - } - } - @Override public JdbcDialect getDialect() { return JdbcDialect.Postgres; @@ -135,29 +106,12 @@ protected PartitionType getPartitionType( } @Override - protected String derivePartitionInterval( + protected Optional derivePartitionInterval( PartitionType partitionType, Duration ttl, ChronoUnit ttlUnit) { if (partitionType != PartitionType.RANGE || ttl == null || ttl.isZero() || ttlUnit == null) { - return null; - } - return derivePartitionInterval(ttl, ttlUnit, partitionDivisor); - } - - /** - * Picks the partition width for a range-partitioned table with a TTL: the TTL divided by the - * configured 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. - */ - static String derivePartitionInterval(Duration ttl, ChronoUnit ttlUnit, int partitionDivisor) { - var floorMinutes = ttlUnit.getDuration().toMinutes(); - var targetMinutes = Math.max(ttl.toMinutes() / (double) partitionDivisor, floorMinutes); - var interval = PARTITION_MENU_LABELS[0]; - for (var i = 0; i < PARTITION_MENU_MINUTES.length; i++) { - if (PARTITION_MENU_MINUTES[i] <= targetMinutes) { - interval = PARTITION_MENU_LABELS[i]; - } + return Optional.empty(); } - return interval; + return Optional.of(PostgresPartitionInterval.of(ttl, ttlUnit, partitionDivisor).getInterval()); } @Override 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 c3d30a2b06..685aba1ef5 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,37 +19,16 @@ 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.Locale; -import java.util.Map; import java.util.Optional; -import java.util.regex.Pattern; -import org.apache.flink.util.TimeUtils; public class TtlHint extends PlannerHint { public static final String HINT_NAME = "ttl"; - private static final Pattern TTL_PATTERN = Pattern.compile("^(\\d+)\\s*([a-zA-Z]+)$"); - - /** Allowed TTL units: at least a minute, at most a week */ - private static final Map TTL_UNITS = - Map.ofEntries( - Map.entry("min", ChronoUnit.MINUTES), - Map.entry("minute", ChronoUnit.MINUTES), - Map.entry("minutes", ChronoUnit.MINUTES), - Map.entry("h", ChronoUnit.HOURS), - Map.entry("hour", ChronoUnit.HOURS), - Map.entry("hours", ChronoUnit.HOURS), - Map.entry("d", ChronoUnit.DAYS), - Map.entry("day", ChronoUnit.DAYS), - Map.entry("days", ChronoUnit.DAYS), - Map.entry("w", ChronoUnit.WEEKS), - Map.entry("week", ChronoUnit.WEEKS), - Map.entry("weeks", ChronoUnit.WEEKS)); - private final Duration ttl; private final ChronoUnit ttlUnit; @@ -94,29 +73,25 @@ public String getName() { } private static TtlHint parseTtlArgument(ParsedObject source, String argument) { - var matcher = TTL_PATTERN.matcher(argument); + Duration ttl = null; ChronoUnit unit = null; - if (matcher.matches()) { - unit = TTL_UNITS.get(matcher.group(2).toLowerCase(Locale.ROOT)); + try { + ttl = TimeUtils.parseDuration(argument); + unit = TimeUtils.parseDurationUnit(argument); + } catch (Exception e) { + // fall through to the shared error below } - if (unit == null) { + 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 a positive number with a" - + " unit between minute and week, e.g. `30 min`, `36 hours`, `14 days`, or `2 weeks`.", + + " unit between minute and day, e.g. `30 min`, `36 hours`, or `14 days`.", source.get().name(), argument); } - var value = Long.parseLong(matcher.group(1)); - var ttl = - switch (unit) { - case MINUTES -> Duration.ofMinutes(value); - case HOURS -> Duration.ofHours(value); - case DAYS -> Duration.ofDays(value); - case WEEKS -> Duration.ofDays(value * 7L); - default -> throw new IllegalStateException("Unexpected TTL unit: " + unit); - }; 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 0000000000..b97e1f0736 --- /dev/null +++ b/sqrl-planner/src/main/java/com/datasqrl/util/TimeUtils.java @@ -0,0 +1,58 @@ +/* + * 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; + +/** Duration parsing utilities built on top of Flink's {@link org.apache.flink.util.TimeUtils}. */ +public final class TimeUtils { + + private 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 a882ad6268..b61afcf856 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-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 5be2a3813b..0f0b0a9ed4 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-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/AbstractJdbcStatementFactoryTest.java b/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactoryTest.java deleted file mode 100644 index c9be05a089..0000000000 --- a/sqrl-planner/src/test/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactoryTest.java +++ /dev/null @@ -1,85 +0,0 @@ -/* - * 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.mockito.Mockito.when; - -import com.datasqrl.error.ErrorCollector; -import com.datasqrl.error.ErrorLocation.FileLocation; -import com.datasqrl.planner.analyzer.TableAnalysis; -import com.datasqrl.planner.hint.PlannerHints; -import com.datasqrl.planner.parser.ParsedObject; -import com.datasqrl.planner.parser.SqrlComments; -import com.datasqrl.planner.parser.SqrlHint; -import com.datasqrl.planner.tables.FlinkTableBuilder; -import com.datasqrl.planner.util.Documented; -import java.time.Duration; -import java.util.List; -import java.util.Optional; -import org.apache.calcite.rel.type.RelDataType; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.Mock; -import org.mockito.junit.jupiter.MockitoExtension; - -@ExtendWith(MockitoExtension.class) -class AbstractJdbcStatementFactoryTest { - - @Mock private TableAnalysis tableAnalysis; - - @Mock private RelDataType datatype; - - @Mock private FlinkTableBuilder tableBuilder; - - private final PostgresStatementFactory factory = new PostgresStatementFactory(); - - private CreateTableJdbcStatement createTable(PlannerHints hints) { - when(tableAnalysis.getHints()).thenReturn(hints); - when(tableAnalysis.getDocumentation()).thenReturn(Documented.EMPTY); - when(datatype.getFieldList()).thenReturn(List.of()); - when(tableBuilder.getPrimaryKey()).thenReturn(Optional.empty()); - - var createTable = new JdbcEngineCreateTable("my_table", tableBuilder, datatype, tableAnalysis); - return (CreateTableJdbcStatement) factory.createTable(createTable); - } - - private static PlannerHints hints(SqrlHint... sqrlHints) { - var parsedHints = - List.of(sqrlHints).stream() - .map(hint -> new ParsedObject<>(hint, FileLocation.START)) - .toList(); - var comments = new SqrlComments(List.of(), parsedHints); - return PlannerHints.from(comments, Optional.empty(), ErrorCollector.root()); - } - - @Test - void givenTtlHint_whenCreateTable_thenTtlPopulated() { - var stmt = createTable(hints(new SqrlHint("ttl", List.of("30 days")))); - - assertThat(stmt.getTtl()).isEqualTo(Duration.ofDays(30)); - // no partition_key hint, so the table is not range-partitioned - assertThat(stmt.getPartitionInterval()).isNull(); - } - - @Test - void givenNoTtlHint_whenCreateTable_thenZeroTtlAndNullInterval() { - var stmt = createTable(PlannerHints.EMPTY); - - assertThat(stmt.getTtl()).isEqualTo(Duration.ZERO); - assertThat(stmt.getPartitionInterval()).isNull(); - } -} 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 0000000000..9c3a2e8cf3 --- /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_whenOf_thenMenuWidth( + long amount, ChronoUnit unit, int divisor, String expected) { + var ttl = Duration.of(amount, unit); + + assertThat(PostgresPartitionInterval.of(ttl, unit, divisor).getInterval()).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 index eac0433283..847209f0e8 100644 --- 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 @@ -20,100 +20,59 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; -import com.datasqrl.config.PackageJson.EmptyEngineConfig; import com.datasqrl.config.PackageJson.EngineConfig; import com.datasqrl.engine.database.relational.CreateTableJdbcStatement.PartitionType; import java.time.Duration; import java.time.temporal.ChronoUnit; -import java.util.Optional; import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.CsvSource; class PostgresStatementFactoryTest { - @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", - "2, WEEKS, 100, 1 week", - // 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_whenDeriveInterval_thenMenuWidth( - long amount, ChronoUnit unit, int divisor, String expected) { - var ttl = unit == ChronoUnit.WEEKS ? Duration.ofDays(amount * 7) : Duration.of(amount, unit); - - assertThat(PostgresStatementFactory.derivePartitionInterval(ttl, unit, divisor)) - .isEqualTo(expected); - } - @Test - void givenNonRangeOrNoTtl_whenDeriveInterval_thenNull() { - var factory = new PostgresStatementFactory(); + void givenNonRangeOrNoTtl_whenDeriveInterval_thenEmpty() { + var factory = new PostgresStatementFactory(100); assertThat( factory.derivePartitionInterval( PartitionType.HASH, Duration.ofDays(14), ChronoUnit.DAYS)) - .isNull(); + .isEmpty(); assertThat(factory.derivePartitionInterval(PartitionType.RANGE, Duration.ZERO, ChronoUnit.DAYS)) - .isNull(); - assertThat(factory.derivePartitionInterval(PartitionType.RANGE, null, null)).isNull(); + .isEmpty(); + assertThat(factory.derivePartitionInterval(PartitionType.RANGE, null, null)).isEmpty(); assertThat(factory.derivePartitionInterval(PartitionType.RANGE, Duration.ofDays(14), null)) - .isNull(); + .isEmpty(); } @Test void givenRangeAndTtl_whenDeriveInterval_thenWidthReturned() { - var factory = new PostgresStatementFactory(); - - assertThat( - factory.derivePartitionInterval( - PartitionType.RANGE, Duration.ofDays(14), ChronoUnit.DAYS)) - .isEqualTo("1 day"); - } - - @Test - void givenEmptyEngineConfig_whenCreate_thenDefaultDivisorUsed() { - var factory = new PostgresStatementFactory(new EmptyEngineConfig("postgres")); + var factory = new PostgresStatementFactory(100); assertThat( factory.derivePartitionInterval( PartitionType.RANGE, Duration.ofDays(14), ChronoUnit.DAYS)) - .isEqualTo("1 day"); + .contains("1 day"); } @Test void givenConfiguredDivisor_whenCreate_thenDivisorApplied() { var engineConfig = mock(EngineConfig.class); - when(engineConfig.getSetting( - PostgresStatementFactory.PARTITION_DIVISOR_KEY, Optional.of("100"))) - .thenReturn("4"); + when(engineConfig.getSetting(PostgresStatementFactory.PARTITION_DIVISOR_KEY)).thenReturn("4"); var factory = new PostgresStatementFactory(engineConfig); assertThat( factory.derivePartitionInterval( PartitionType.RANGE, Duration.ofDays(84), ChronoUnit.DAYS)) - .isEqualTo("2 weeks"); + .contains("2 weeks"); } @Test void givenInvalidDivisor_whenCreate_thenThrows() { var engineConfig = mock(EngineConfig.class); - when(engineConfig.getSetting( - PostgresStatementFactory.PARTITION_DIVISOR_KEY, Optional.of("100"))) + when(engineConfig.getSetting(PostgresStatementFactory.PARTITION_DIVISOR_KEY)) .thenReturn("not-a-number"); assertThatThrownBy(() -> new PostgresStatementFactory(engineConfig)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("partition-divisor"); + .isInstanceOf(NumberFormatException.class); assertThatThrownBy(() -> new PostgresStatementFactory(0)) .isInstanceOf(IllegalArgumentException.class) 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 index cc01ef27aa..d83f43fbdc 100644 --- a/sqrl-planner/src/test/java/com/datasqrl/planner/hint/TtlHintTest.java +++ b/sqrl-planner/src/test/java/com/datasqrl/planner/hint/TtlHintTest.java @@ -64,8 +64,6 @@ void givenNoArgs_whenCreate_thenEmptyTtlAndUnit() { "1 h, 60, HOURS", "14 days, 20160, DAYS", "1 d, 1440, DAYS", - "2 weeks, 20160, WEEKS", - "5 week, 50400, WEEKS", "14days, 20160, DAYS" }) void givenSupportedUnit_whenCreate_thenTtlAndUnitParsed( @@ -78,11 +76,22 @@ void givenSupportedUnit_whenCreate_thenTtlAndUnitParsed( @ParameterizedTest @ValueSource( - strings = {"10 s", "10 seconds", "500 ms", "3 months", "1 year", "fortnight", "14", "days"}) + 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 week"); + .hasMessageContaining("unit between minute and day"); } @Test 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 0000000000..6e34872406 --- /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); + } +} From acd011b4f03f4e3ac5b7805c92044201715eea01 Mon Sep 17 00:00:00 2001 From: Wellington Mafra Date: Wed, 29 Jul 2026 14:04:57 -0300 Subject: [PATCH 6/8] test: Fix merge fallout from deployment model extraction Signed-off-by: Wellington Mafra --- .../com/datasqrl/planner/PlanModelSerializationTest.java | 1 + .../snapshots/com/datasqrl/DAGPlannerTest/partitionTest.txt | 6 ++---- 2 files changed, 3 insertions(+), 4 deletions(-) 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 e52867f42d..a2ae884e50 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-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 9d7ad8bdd3..ade4ef2a98 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 @@ -246,8 +246,7 @@ END ], "partitionType" : "RANGE", "numPartitions" : 1, - "ttl" : 172800.000000000, - "partitionInterval" : "1 hour" + "ttl" : 172800.000000000 }, { "name" : "OrderUpdates_2", @@ -284,8 +283,7 @@ END ], "partitionType" : "RANGE", "numPartitions" : 1, - "ttl" : 2592000.000000000, - "partitionInterval" : "1 day" + "ttl" : 2592000.000000000 }, { "name" : "Orders_3", From 156812d747b37867c69886d5c34d45d22c7ea384 Mon Sep 17 00:00:00 2001 From: Wellington Mafra Date: Wed, 29 Jul 2026 17:18:02 -0300 Subject: [PATCH 7/8] refactor: Rename setting to partition-ttl-divisor Signed-off-by: Wellington Mafra --- .../docs/configuration-engine/postgres.md | 10 +++++----- .../relational/PostgresPartitionInterval.java | 4 ++-- .../relational/PostgresStatementFactory.java | 16 +++++++++------- .../src/main/resources/default-package.json | 2 +- .../main/resources/jsonSchema/packageSchema.json | 2 +- .../relational/PostgresStatementFactoryTest.java | 7 ++++--- 6 files changed, 22 insertions(+), 19 deletions(-) diff --git a/documentation/docs/configuration-engine/postgres.md b/documentation/docs/configuration-engine/postgres.md index 4419fe4c53..9dbae4e0ef 100644 --- a/documentation/docs/configuration-engine/postgres.md +++ b/documentation/docs/configuration-engine/postgres.md @@ -6,9 +6,9 @@ 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-divisor` | **integer** | `100` | Controls the number of partitions for range-partitioned tables with a TTL (see [Partitioning](#partitioning)). | +| 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 @@ -16,7 +16,7 @@ No mandatory configuration keys are required. Physical DDL (tables, indexes, vie { "engines": { "postgres": { - "partition-divisor": 100, + "partition-ttl-divisor": 100, "config": { // Optional PostgreSQL-specific settings } @@ -29,7 +29,7 @@ No mandatory configuration keys are required. Physical DDL (tables, indexes, vie 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-divisor` caps the number of partitions. +- 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. 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 index a9e071a077..248e3ff039 100644 --- 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 @@ -54,9 +54,9 @@ public class PostgresPartitionInterval { * the closest calendar-aligned width from the menu. */ public static PostgresPartitionInterval of( - Duration ttl, ChronoUnit ttlUnit, int partitionDivisor) { + Duration ttl, ChronoUnit ttlUnit, int partitionTtlDivisor) { var floorMinutes = ttlUnit.getDuration().toMinutes(); - var targetMinutes = Math.max(ttl.toMinutes() / (double) partitionDivisor, floorMinutes); + var targetMinutes = Math.max(ttl.toMinutes() / (double) partitionTtlDivisor, floorMinutes); var interval = PARTITION_MENU.values().iterator().next(); for (var entry : PARTITION_MENU.entrySet()) { if (entry.getKey() <= targetMinutes) { 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 021f0ccfd4..d00062f38d 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 @@ -49,18 +49,19 @@ public class PostgresStatementFactory extends AbstractJdbcStatementFactory { - public static final String PARTITION_DIVISOR_KEY = "partition-divisor"; + public static final String PARTITION_TTL_DIVISOR_KEY = "partition-ttl-divisor"; - private final int partitionDivisor; + private final int partitionTtlDivisor; public PostgresStatementFactory(EngineConfig engineConfig) { - this(Integer.parseInt(engineConfig.getSetting(PARTITION_DIVISOR_KEY))); + this(Integer.parseInt(engineConfig.getSetting(PARTITION_TTL_DIVISOR_KEY))); } - public PostgresStatementFactory(int partitionDivisor) { + public PostgresStatementFactory(int partitionTtlDivisor) { super(Dialect.POSTGRES, new PostgresCreateTableDdlFactory(true)); - checkArgument(partitionDivisor > 0, "%s must be a positive number", PARTITION_DIVISOR_KEY); - this.partitionDivisor = partitionDivisor; + checkArgument( + partitionTtlDivisor > 0, "%s must be a positive number", PARTITION_TTL_DIVISOR_KEY); + this.partitionTtlDivisor = partitionTtlDivisor; } @Override @@ -111,7 +112,8 @@ protected Optional derivePartitionInterval( if (partitionType != PartitionType.RANGE || ttl == null || ttl.isZero() || ttlUnit == null) { return Optional.empty(); } - return Optional.of(PostgresPartitionInterval.of(ttl, ttlUnit, partitionDivisor).getInterval()); + return Optional.of( + PostgresPartitionInterval.of(ttl, ttlUnit, partitionTtlDivisor).getInterval()); } @Override diff --git a/sqrl-planner/src/main/resources/default-package.json b/sqrl-planner/src/main/resources/default-package.json index b61afcf856..282c979e9c 100644 --- a/sqrl-planner/src/main/resources/default-package.json +++ b/sqrl-planner/src/main/resources/default-package.json @@ -34,7 +34,7 @@ "url": "jdbc:duckdb:" }, "postgres": { - "partition-divisor": 100 + "partition-ttl-divisor": 100 }, "kafka": { "retention": null, diff --git a/sqrl-planner/src/main/resources/jsonSchema/packageSchema.json b/sqrl-planner/src/main/resources/jsonSchema/packageSchema.json index 0f0b0a9ed4..a5c0e3f6d7 100644 --- a/sqrl-planner/src/main/resources/jsonSchema/packageSchema.json +++ b/sqrl-planner/src/main/resources/jsonSchema/packageSchema.json @@ -265,7 +265,7 @@ "postgres": { "type": "object", "properties": { - "partition-divisor": { + "partition-ttl-divisor": { "description": "Divisor applied to a table's TTL to cap the number of pg_partman partitions.", "type": "integer" }, 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 index 0c4961204a..2666beade5 100644 --- 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 @@ -56,7 +56,8 @@ void givenRangeAndTtl_whenDeriveInterval_thenWidthReturned() { @Test void givenConfiguredDivisor_whenCreate_thenDivisorApplied() { var engineConfig = mock(EngineConfig.class); - when(engineConfig.getSetting(PostgresStatementFactory.PARTITION_DIVISOR_KEY)).thenReturn("4"); + when(engineConfig.getSetting(PostgresStatementFactory.PARTITION_TTL_DIVISOR_KEY)) + .thenReturn("4"); var factory = new PostgresStatementFactory(engineConfig); assertThat( @@ -68,7 +69,7 @@ void givenConfiguredDivisor_whenCreate_thenDivisorApplied() { @Test void givenInvalidDivisor_whenCreate_thenThrows() { var engineConfig = mock(EngineConfig.class); - when(engineConfig.getSetting(PostgresStatementFactory.PARTITION_DIVISOR_KEY)) + when(engineConfig.getSetting(PostgresStatementFactory.PARTITION_TTL_DIVISOR_KEY)) .thenReturn("not-a-number"); assertThatThrownBy(() -> new PostgresStatementFactory(engineConfig)) @@ -76,6 +77,6 @@ void givenInvalidDivisor_whenCreate_thenThrows() { assertThatThrownBy(() -> new PostgresStatementFactory(0)) .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("partition-divisor"); + .hasMessageContaining("partition-ttl-divisor"); } } From d29694b0f9cfbd5f79fee82144b1b4e810b0dd3a Mon Sep 17 00:00:00 2001 From: Wellington Mafra Date: Thu, 30 Jul 2026 11:22:03 -0300 Subject: [PATCH 8/8] refactor: Address final pg_partman review comments Signed-off-by: Wellington Mafra --- .../relational/PostgresPartitionInterval.java | 22 +++++++------- .../relational/PostgresStatementFactory.java | 3 +- .../com/datasqrl/planner/hint/CacheHint.java | 30 +++++++++++++++++-- .../com/datasqrl/planner/hint/TtlHint.java | 29 ------------------ .../java/com/datasqrl/util/TimeUtils.java | 5 ++-- .../PostgresPartitionIntervalTest.java | 4 +-- 6 files changed, 45 insertions(+), 48 deletions(-) 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 index 248e3ff039..0e37958ec0 100644 --- 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 @@ -19,11 +19,12 @@ import java.time.Duration; import java.time.temporal.ChronoUnit; import java.util.Map; -import lombok.Value; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; -/** The pg_partman partition width for a range-partitioned table with a TTL. */ -@Value -public class PostgresPartitionInterval { +/** 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 PARTITION_MENU = @@ -46,23 +47,22 @@ public class PostgresPartitionInterval { .put(120960, "12 weeks") .buildOrThrow(); - String interval; - /** * 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 PostgresPartitionInterval of( - Duration ttl, ChronoUnit ttlUnit, int partitionTtlDivisor) { + public static String asString(Duration ttl, ChronoUnit ttlUnit, int partitionTtlDivisor) { var floorMinutes = ttlUnit.getDuration().toMinutes(); var targetMinutes = Math.max(ttl.toMinutes() / (double) partitionTtlDivisor, floorMinutes); - var interval = PARTITION_MENU.values().iterator().next(); + + String interval = null; for (var entry : PARTITION_MENU.entrySet()) { - if (entry.getKey() <= targetMinutes) { + if (interval == null || entry.getKey() <= targetMinutes) { interval = entry.getValue(); } } - return new PostgresPartitionInterval(interval); + + 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 d00062f38d..8e2b3e5302 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 @@ -112,8 +112,7 @@ protected Optional derivePartitionInterval( if (partitionType != PartitionType.RANGE || ttl == null || ttl.isZero() || ttlUnit == null) { return Optional.empty(); } - return Optional.of( - PostgresPartitionInterval.of(ttl, ttlUnit, partitionTtlDivisor).getInterval()); + return Optional.of(PostgresPartitionInterval.asString(ttl, ttlUnit, partitionTtlDivisor)); } @Override 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 e22536010e..dc1a01de89 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 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 685aba1ef5..f7eda98bef 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 @@ -94,33 +94,4 @@ private static TtlHint parseTtlArgument(ParsedObject source, String ar } return new TtlHint(source, ttl, unit); } - - public static Duration parseDuration(ParsedObject 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()); - } - return parseDurationArgument(source, arguments.get(0)); - } - - private static Duration parseDurationArgument(ParsedObject source, String argument) { - try { - return TimeUtils.parseDuration(argument); - } 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(), - argument); - } - } } diff --git a/sqrl-planner/src/main/java/com/datasqrl/util/TimeUtils.java b/sqrl-planner/src/main/java/com/datasqrl/util/TimeUtils.java index b97e1f0736..e48667471d 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/util/TimeUtils.java +++ b/sqrl-planner/src/main/java/com/datasqrl/util/TimeUtils.java @@ -18,12 +18,13 @@ 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 { - private 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 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 index 9c3a2e8cf3..ea5553f29e 100644 --- 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 @@ -38,10 +38,10 @@ class PostgresPartitionIntervalTest { // target below the smallest menu entry falls back to the smallest "30, MINUTES, 100, 15 minutes" }) - void givenTtlAndDivisor_whenOf_thenMenuWidth( + void givenTtlAndDivisor_whenAsString_thenMenuWidth( long amount, ChronoUnit unit, int divisor, String expected) { var ttl = Duration.of(amount, unit); - assertThat(PostgresPartitionInterval.of(ttl, unit, divisor).getInterval()).isEqualTo(expected); + assertThat(PostgresPartitionInterval.asString(ttl, unit, divisor)).isEqualTo(expected); } }