[FLINK-38532] make small changes to improve code readability

Author: raminqaf

docs/content/docs/dev/table/materialized-table/statements.md

@@ -223,8 +223,7 @@ CREATE MATERIALIZED TABLE my_materialized_table_continuous
         'partition.fields.ds.date-formatter' = 'yyyy-MM-dd'
     )
     FRESHNESS = INTERVAL '10' SECOND
-    AS
-    SELECT
+    AS SELECT
         k.ds,
         k.user_id,
         COUNT(*) AS event_count,

flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

@@ -259,7 +259,6 @@ private void createMaterializedTableInFullMode(
         ResolvedCatalogMaterializedTable catalogMaterializedTable =
                 createMaterializedTableOperation.getCatalogMaterializedTable();
 
-        // convert duration to cron expression
         final IntervalFreshness freshness = catalogMaterializedTable.getDefinitionFreshness();
         String cronExpression = convertFreshnessToCron(freshness);
         // create full refresh job

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java

@@ -321,7 +321,7 @@ static String extractFreshness(ResolvedCatalogMaterializedTable materializedTabl
         final IntervalFreshness definitionFreshness = materializedTable.getDefinitionFreshness();
         return String.format(
                 "FRESHNESS = INTERVAL '%s' %s",
-                definitionFreshness.getIntervalInt(), definitionFreshness.getTimeUnit());
+                definitionFreshness.getInterval(), definitionFreshness.getTimeUnit());
     }
 
     static String extractRefreshMode(ResolvedCatalogMaterializedTable materializedTable) {

flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/IntervalFreshness.java

@@ -22,6 +22,7 @@
 import org.apache.flink.table.api.ValidationException;
 
 import java.time.Duration;
+import java.time.temporal.ChronoUnit;
 import java.util.Objects;
 
 /**
@@ -116,24 +117,17 @@ public Duration toDuration() {
      * Prefers larger units when possible (e.g., 60 seconds → 1 minute).
      */
     public static IntervalFreshness fromDuration(Duration duration) {
-        long totalSeconds = duration.getSeconds();
-
-        long days = duration.toDays();
-        if (days * 24 * 60 * 60 == totalSeconds) {
-            return IntervalFreshness.ofDay(String.valueOf(days));
+        if (duration.equals(duration.truncatedTo(ChronoUnit.DAYS))) {
+            return IntervalFreshness.ofDay(String.valueOf(duration.toDays()));
         }
-
-        long hours = duration.toHours();
-        if (hours * 60 * 60 == totalSeconds) {
-            return IntervalFreshness.ofHour(String.valueOf(hours));
+        if (duration.equals(duration.truncatedTo(ChronoUnit.HOURS))) {
+            return IntervalFreshness.ofHour(String.valueOf(duration.toHours()));
         }
-
-        long minutes = duration.toMinutes();
-        if (minutes * 60 == totalSeconds) {
-            return IntervalFreshness.ofMinute(String.valueOf(minutes));
+        if (duration.equals(duration.truncatedTo(ChronoUnit.MINUTES))) {
+            return IntervalFreshness.ofMinute(String.valueOf(duration.toMinutes()));
         }
 
-        return IntervalFreshness.ofSecond(String.valueOf(totalSeconds));
+        return IntervalFreshness.ofSecond(String.valueOf(duration.getSeconds()));
     }
 
     @Override

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java

@@ -130,9 +130,7 @@ void testCreateMaterializedTable() {
         CreateMaterializedTableOperation op = (CreateMaterializedTableOperation) operation;
         ResolvedCatalogMaterializedTable materializedTable = op.getCatalogMaterializedTable();
 
-        Map<String, String> options = new HashMap<>();
-        options.put("connector", "filesystem");
-        options.put("format", "json");
+        Map<String, String> options = Map.of("connector", "filesystem", "format", "json");
         CatalogMaterializedTable expected =
                 CatalogMaterializedTable.newBuilder()
                         .schema(
@@ -185,9 +183,7 @@ void testCreateMaterializedTableWithoutFreshness() {
         ResolvedCatalogMaterializedTable materializedTable = op.getCatalogMaterializedTable();
         assertThat(materializedTable).isInstanceOf(ResolvedCatalogMaterializedTable.class);
 
-        Map<String, String> options = new HashMap<>();
-        options.put("connector", "filesystem");
-        options.put("format", "json");
+        Map<String, String> options = Map.of("connector", "filesystem", "format", "json");
 
         CatalogMaterializedTable expected =
                 CatalogMaterializedTable.newBuilder()
@@ -241,9 +237,7 @@ void testCreateMaterializedTableWithoutFreshnessAndRefreshMode() {
         ResolvedCatalogMaterializedTable materializedTable = op.getCatalogMaterializedTable();
         assertThat(materializedTable).isInstanceOf(ResolvedCatalogMaterializedTable.class);
 
-        Map<String, String> options = new HashMap<>();
-        options.put("connector", "filesystem");
-        options.put("format", "json");
+        Map<String, String> options = Map.of("connector", "filesystem", "format", "json");
 
         CatalogMaterializedTable expected =
                 CatalogMaterializedTable.newBuilder()