[Bug] Config index-ttl for Cross Partitions Upsert will result RocksDBException(value's length less than timestamp's) in bootstrap stage

[linhao1990]

Search before asking

• [X] I searched in the issues and found nothing similar.

Paimon version

0.7

Compute Engine

Flink 1.18

Minimal reproduce step

org.apache.paimon.crosspartition.GlobalIndexAssignerTest
    @Test
    public void testBootstrapRecords() throws Exception {
        // enableTtl is true
        GlobalIndexAssigner assigner = createAssigner(MergeEngine.DEDUPLICATE, true); 
        List<List<Integer>> output = new ArrayList<>();
        assigner.open(
                0,
                ioManager(),
                2,
                0,
                (row, bucket) ->
                        output.add(
                                Arrays.asList(
                                        row.getInt(0), row.getInt(1), row.getInt(2), bucket)));

        // assigner.bootstrapKey can trigger the problem
        assigner.bootstrapKey(GenericRow.of(1, 1, 1)); 

        assigner.processInput(GenericRow.of(1, 1, 1));
        assigner.endBoostrap(true);
}

What doesn't meet your expectations?

I think the bootstrap should finish, but it throw the following exception:

org.rocksdb.RocksDBException: Error: value's length less than timestamp's

	at org.apache.paimon.lookup.RocksDBValueState.get(RocksDBValueState.java:49)
	at org.apache.paimon.crosspartition.GlobalIndexAssigner.processInput(GlobalIndexAssigner.java:238)
	at org.apache.paimon.crosspartition.GlobalIndexAssigner.loopBootstrapRecords(GlobalIndexAssigner.java:365)
	at org.apache.paimon.crosspartition.GlobalIndexAssigner.endBoostrap(GlobalIndexAssigner.java:223)
	at org.apache.paimon.crosspartition.GlobalIndexAssignerTest.testBootstrapRecords(GlobalIndexAssignerTest.java:236)

I think the error is related to the following code part, paimon uses TtlDB implementation.

org.apache.paimon.lookup.RocksDBStateFactory#RocksDBStateFactory
this.db =
        ttlSecs == null
                ? RocksDB.open(options, path)
                : TtlDB.open(options, path, (int) ttlSecs.getSeconds(), false);

In rocksdb I find the code throws the error:

// Returns corruption if the length of the string is lesser than timestamp, or
// timestamp refers to a time lesser than ttl-feature release time
Status DBWithTTLImpl::SanityCheckTimestamp(const Slice& str) {
  if (str.size() < kTSLength) {
    return Status::Corruption("Error: value's length less than timestamp's\n");
  }
}

I do not found how we set timestamp when add record into rocksdb.

Anything else?

No response

Are you willing to submit a PR?

• [ ] I'm willing to submit a PR!

[JingsongLi]

thanks @linhao1990 for reporting. this is because [SST files written using SstFileWriter don't work with TtlDB](https://github.com/facebook/rocksdb/issues/7299).