[core] Support seconds-long in record level time field type (apache#4133

)
xiangyuf · Sep 5, 2024 · aa04090 · aa04090
1 parent 8193b86
commit aa04090
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 7 deletions.
diff --git a/docs/content/primary-key-table/compaction.md b/docs/content/primary-key-table/compaction.md
@@ -76,7 +76,7 @@ In compaction, you can configure record-Level expire time to expire records, you
 
 1. `'record-level.expire-time'`: time retain for records.
 2. `'record-level.time-field'`: time field for record level expire.
-3. `'record-level.time-field-type'`: time field type for record level expire, it can be seconds-int or millis-long.
+3. `'record-level.time-field-type'`: time field type for record level expire, it can be seconds-int,seconds-long or millis-long.
 
 Expiration happens in compaction, and there is no strong guarantee to expire records in time.
 

diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -591,7 +591,7 @@
             <td><h5>record-level.time-field-type</h5></td>
             <td style="word-wrap: break-word;">seconds-int</td>
             <td><p>Enum</p></td>
-            <td>Time field type for record level expire, it can be seconds-int or millis-long.<br /><br />Possible values:<ul><li>"seconds-int": Timestamps in seconds should be INT type.</li><li>"millis-long": Timestamps in milliseconds should be BIGINT type.</li></ul></td>
+            <td>Time field type for record level expire, it can be seconds-int,seconds-long or millis-long.<br /><br />Possible values:<ul><li>"seconds-int": Timestamps in seconds with INT field type.</li><li>"seconds-long": Timestamps in seconds with BIGINT field type.</li><li>"millis-long": Timestamps in milliseconds with BIGINT field type.</li></ul></td>
         </tr>
         <tr>
             <td><h5>rowkind.field</h5></td>

diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1281,7 +1281,7 @@ public class CoreOptions implements Serializable {
                     .enumType(TimeFieldType.class)
                     .defaultValue(TimeFieldType.SECONDS_INT)
                     .withDescription(
-                            "Time field type for record level expire, it can be seconds-int or millis-long.");
+                            "Time field type for record level expire, it can be seconds-int,seconds-long or millis-long.");
 
     public static final ConfigOption<String> FIELDS_DEFAULT_AGG_FUNC =
             key(FIELDS_PREFIX + "." + DEFAULT_AGG_FUNCTION)
@@ -2728,9 +2728,11 @@ public InlineElement getDescription() {
 
     /** Time field type for record level expire. */
     public enum TimeFieldType implements DescribedEnum {
-        SECONDS_INT("seconds-int", "Timestamps in seconds should be INT type."),
+        SECONDS_INT("seconds-int", "Timestamps in seconds with INT field type."),
 
-        MILLIS_LONG("millis-long", "Timestamps in milliseconds should be BIGINT type.");
+        SECONDS_LONG("seconds-long", "Timestamps in seconds with BIGINT field type."),
+
+        MILLIS_LONG("millis-long", "Timestamps in milliseconds with BIGINT field type.");
 
         private final String value;
         private final String description;

diff --git a/paimon-core/src/main/java/org/apache/paimon/io/RecordLevelExpire.java b/paimon-core/src/main/java/org/apache/paimon/io/RecordLevelExpire.java
@@ -64,12 +64,15 @@ public static RecordLevelExpire create(CoreOptions options, RowType rowType) {
         DataField field = rowType.getField(timeField);
         if (!((timeFieldType == CoreOptions.TimeFieldType.SECONDS_INT
                         && field.type() instanceof IntType)
+                || (timeFieldType == CoreOptions.TimeFieldType.SECONDS_LONG
+                        && field.type() instanceof BigIntType)
                 || (timeFieldType == CoreOptions.TimeFieldType.MILLIS_LONG
                         && field.type() instanceof BigIntType))) {
             throw new IllegalArgumentException(
                     String.format(
-                            "Record level time field should be INT type, but is %s.",
-                            field.type()));
+                            "The record level time field type should be one of SECONDS_INT,SECONDS_LONG or MILLIS_LONG, "
+                                    + "but time field type is %s, field type is %s.",
+                            timeFieldType, field.type()));
         }
 
         return new RecordLevelExpire(fieldIndex, (int) expireTime.getSeconds(), timeFieldType);
@@ -98,6 +101,9 @@ private RecordReader<KeyValue> wrap(RecordReader<KeyValue> reader) {
                         case SECONDS_INT:
                             recordTime = kv.value().getInt(timeField);
                             break;
+                        case SECONDS_LONG:
+                            recordTime = (int) kv.value().getLong(timeField);
+                            break;
                         case MILLIS_LONG:
                             recordTime = (int) (kv.value().getLong(timeField) / 1000);
                             break;