Prune checkpoints in Lambda

[rdettai]

Description

Closes #4613

Avoid accumulating file sources when running Lambda indexer

How was this PR tested?

Describe how you tested this PR.

[rdettai]

A bug remains, only one source is created:

    "checkpoint": {
      "ingest-lambda-source-1711043830": {
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711043826.gz": "00000000000016044634",
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711044126.gz": "00000000000016043012",
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711044426.gz": "00000000000016038941",
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711044726.gz": "00000000000016041053",
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711045026.gz": "00000000000016041903",
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711045326.gz": "00000000000016044080",
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711045626.gz": "00000000000016041526",
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711045925.gz": "00000000000016042481",
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711046225.gz": "00000000000016044227",
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711046526.gz": "00000000000016042532",
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711046825.gz": "00000000000016043210",
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711047126.gz": "00000000000016042689",
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711047426.gz": "00000000000016042582",
        "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711047726.gz": "00000000000016042450"
      }
    },
    "create_timestamp": 1711039331,
    "sources": [
      {
        "version": "0.8",
        "source_id": "ingest-lambda-source-1711043830",
        "num_pipelines": 1,
        "enabled": true,
        "source_type": "file",
        "params": {
          "filepath": "s3://mockdatastack-sourcemockdata26422bfc-mpua0jb4rrh1/mock-sales/1711043826.gz"
        },
        "input_format": "json"
      }
    ]
  },

[trinity-1686a]

i'd rather we always keep the last X checkpoints so as to make sure we are resilient to redundant notification, but i'm not sure how that can be done (embed a timestamp in the partition id, after a # maybe?). Anyway that's definitely an improvement

[rdettai]

i'd rather we always keep the last X checkpoints

Yes, couldn't agree more. I logged in #4613 a few things I tried to achieve that and why they failed. We could definitively come up with a solution but it would require a more massive rewrite on the metastore or the source.