...

-- By default, if no time is specified, the system will query data from the latest timestamp. 

SELECT [column_name(s)] FROM [table_name] 

-- For example: SELECT * FROM paimon_tb;



-- Specify a current-time expression 

SELECT [column_name(s)] FROM [table_name] FOR SYSTEM_TIME AS OF [current_time expression];

-- For example: SELECT * FROM paimon_tb FOR SYSTEM_TIME AS OF CURRENT_TIMESTAMP;

Timestamp Expression

Supported Timestamp type

As shown in the above syntax, when querying with time travel, we support specifying a constant expression of time. Therefore, the type corresponding to this expression may be TIMESTAMP or TIMESTAMP_LTZ.

But the interface reserved for the Connector layer is a long value timestamp(The newly added method for the Catalog is as follows) which means that we need to convert both TIMESTAMP and TIMESTAMP_LTZ expressions into long values in the end.

By default, it is natural to convert data of type TIMESTAMP_LTZ to Long value. However, if it is a TIMESTAMP type, we will implicitly convert it to TIMESTAMP_LTZ and then convert it to a Long value.

Supported Expression Syntax

<TIMESTAMP EXPRESSION> must be a valid and executable expression.  

For example, we will support TIMESTAMP '2023-05-05 00:00:00', but not support TIMESTAMP_LTZ '2023-05-05 00:00:00'. If you need to specify the TIMESTAMP_LTZ type, you need to use TO_TIMESTAMP_LTZ(1, 3).

Public interfaces

Catalog

@PublicEvolving
public interface Catalog {

   /**
     * Returns a {@link CatalogTable} or {@link CatalogView} identified by the given {@link
     * ObjectPath}. The framework will resolve the metadata objects when necessary.
     *
     * @param tablePath Path of the table or view
     * @param timestamp Timestamp of the table snapshot, which is milliseconds since 1970-01-01 00:00:00 UTC 
     * @return The requested table or view
     * @throws TableNotExistException if the target does not exist
     * @throws CatalogException in case of any runtime exception
     */
    default CatalogBaseTable getTable(ObjectPath tablePath, long timestamp)
            throws TableNotExistException, CatalogException {
        throw new UnsupportedOperationException(
                String.format("Table %s does not support time travel.", tablePath));
    }

}

...

@PublicEvolving
public interface CatalogTable extends CatalogBaseTable {
    /** Return the snapshot specified for the table. Return Optional.empty() if not specified. */
    default Optional<Long> getSnapshot() {
        return Optional.empty();
    }
}

Supported Time Type

As shown in the above syntax, when querying with time travel, we support specifying a constant expression of time. Therefore, the type corresponding to this expression may be TIMESTAMP or TIMESTAMP_LTZ.

But the interface reserved for the Connector layer is a long value timestamp, which means that we need to convert both TIMESTAMP and TIMESTAMP_LTZ expressions into long values in the end.

...

Proposed Changes

Flink SQL related Changes

...