-- To specify a specific time, use the following syntax: 

SELECT [column_name(s)] FROM [table_name] FOR SYSTEM_TIME AS OF TIMESTAMP '[timestamp]'

-- For example: SELECT * FROM paimon_tb FOR SYSTEM_TIME AS OF TIMESTAMP '2023-04-27 00:00:00'


-- To specify a constant expression, use the following syntax: 

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

-- For example: SELECT * FROM paimon_tb FOR SYSTEM_TIME AS OF TIMESTAMP '2023-04-27 00:00:00' - INTERVAL '1' DAY

-- 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;

@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();
    }
}

```
@Override  
public DynamicTableSource createDynamicTableSource(Context context) {   
  
Optional<Long> snapshot = context.getCatalogTable().getSnapshot();
```

/** Catalog for paimon. */
public class FlinkCatalog extends AbstractCatalog {
		
	...

    @Override
    public CatalogTable getTable(ObjectPath tablePath)
            throws TableNotExistException, CatalogException {
        Table table;
        try {
            table = catalog.getTable(toIdentifier(tablePath));
        } catch (Catalog.TableNotExistException e) {
            throw new TableNotExistException(getName(), tablePath);
        }

        if (table instanceof FileStoreTable) {
            return toCatalogTable(table);
        } else {
            return new SystemCatalogTable(table);
        }
    }
	
    @Override
    public CatalogTable getTable(ObjectPath tablePath, long timestamp)
            throws TableNotExistException, CatalogException {
		// Due to Paimon's Catalog not providing an interface for historical Schema,
		//  we can only obtain the latest Schema at present.
        CatalogTable catalogTable = this.getTable(tablePath);

		// Attach the specified consumption time parameter.
        Options option =
                new Options()
                        .set(CoreOptions.SCAN_TIMESTAMP_MILLIS, timestamp);

        return catalogTable.copy(option.toMap());
    }

	....
}

SqlSnapshot Snapshot(SqlNode tableRef) :
{
    final Span s;
    final SqlNode e;
}
{
    { s = span(); } <FOR> <SYSTEM_TIME> <AS> <OF>
    // Syntax for temporal table in
    // standard SQL 2011 IWD 9075-2:201?(E) 7.6 <table reference>
    // supports grammar as following:
    // 1. datetime literal
    // 2. datetime value function, i.e. CURRENT_TIMESTAMP
    // 3. datetime term in 1 or 2 +(or -) interval term

    // We extend to support column reference, use Expression
    // to simplify the parsing code.
    e = Expression(ExprContext.ACCEPT_NON_QUERY) {
        return new SqlSnapshot(s.end(this), tableRef, e);
    }
}

SELECT ENo,EName,Sys_Start,Sys_End
FROM Emp FOR SYSTEM_TIME FROM
 TIMESTAMP '2011-01-02 00:00:00’TO
 TIMESTAMP '2011-12-31 00:00:00' 

SELECT ENo,EName,Sys_Start,Sys_End
FROM Emp FOR SYSTEM_TIME BETWEEN
 TIMESTAMP '2011-01-02 00:00:00'AND
 TIMESTAMP '2011-12-31 00:00:00' 

/**
 Enables to push down the timestamp of snapshot into a {@link ScanTableSource}.
 *
 * <p>When a source connector implements the SupportsTimeTravel interface, users can leverage its
 * time-travel functionality to analyze historical data, perform backfilling operations, and more.
 * This can be particularly useful for debugging and data recovery purposes.
 */

@PublicEvolving
public interface SupportsTimeTravel {

    /**
     * @param timestamp The timestamp of the snapshot to apply.
     */
    void applySnapshot(long timestamp);
}