Sample code + merging + tombstone cleanup seems to misbehave

[tarasglek]

I added:

• some logging to your sample code to try to understand merging and tombstone cleanup logic.
• added a merge call. Was a little confused by why alive files kept growing until I realized the merge is one partition at a time(unlike insert). You cover this in docs, but maybe good to mention this in pydoc too.
• added a tombstone call to remove data, logs that shouldn't be needed after merge

Somehow that results in files that were deleted still being in alive files needed by duckdb on 3rd re-run(eg after s3 cleanup). This kind of failure currently seems impossible to debug with stock logs, unless maybe you can infer this from logs. Sorry if this is some newb mistake, still trying to figure out details(eg what _m in name does, log priority) to make a mental picture of how merging works.

(icedb) taras@b550:~/Downloads/icedb$ docker compose up -d
[+] Running 2/4
 ⠦ Network backing-infra_default            Created                                                                                                                                                                                                                        0.6s
 ⠴ Volume "backing-infra_minio_storage"     Created                                                                                                                                                                                                                        0.5s
 ✔ Container minio                          Started                                                                                                                                                                                                                        0.3s
 ✔ Container backing-infra-createbuckets-1  Started                                                                                                                                                                                                                        0.5s
(icedb) taras@b550:~/Downloads/icedb$ python taras/sample.py ;mc ls myminio/testbucket/example/_log/
2 inserted files: ['example/_data/u=user_a/d=2023-06-07/103228e7-d363-4ec9-b9b9-fd965a06bdab.parquet', 'example/_data/u=user_b/d=2023-02-11/626680b5-50a5-4934-b640-acd495a35377.parquet']
2 alive files: ['example/_data/u=user_a/d=2023-06-07/103228e7-d363-4ec9-b9b9-fd965a06bdab.parquet', 'example/_data/u=user_b/d=2023-02-11/626680b5-50a5-4934-b640-acd495a35377.parquet']
0 tombstoned files: []
┌─────────┬──────────────┬──────────┐
│ user_id │ count_star() │   page   │
│ varchar │    int64     │ varchar  │
├─────────┼──────────────┼──────────┤
│ user_a  │            2 │ Home     │
│ user_a  │            1 │ Settings │
│ user_b  │            1 │ Home     │
└─────────┴──────────────┴──────────┘

Keeping 0 files
0 cleaned log files: []
0 deleted log files: []
0 deleted data files: []
[2024-12-22 14:44:19 EET]   388B STANDARD 1734871459837_dan-mbp.jsonl
[2024-12-22 14:44:19 EET]    49B STANDARD 1734871459880_m_dan-mbp.jsonl
(icedb) taras@b550:~/Downloads/icedb$ python taras/sample.py ;mc ls myminio/testbucket/example/_log/
2 inserted files: ['example/_data/u=user_a/d=2023-06-07/b72d0af4-6f1f-40f4-b646-a263f199abe6.parquet', 'example/_data/u=user_b/d=2023-02-11/38fabb73-acc0-489a-9d03-9e10391ca8a7.parquet']
4 alive files: ['example/_data/u=user_a/d=2023-06-07/103228e7-d363-4ec9-b9b9-fd965a06bdab.parquet', 'example/_data/u=user_b/d=2023-02-11/626680b5-50a5-4934-b640-acd495a35377.parquet', 'example/_data/u=user_a/d=2023-06-07/b72d0af4-6f1f-40f4-b646-a263f199abe6.parquet', 'exa
mple/_data/u=user_b/d=2023-02-11/38fabb73-acc0-489a-9d03-9e10391ca8a7.parquet']
0 tombstoned files: []
┌─────────┬──────────────┬──────────┐
│ user_id │ count_star() │   page   │
│ varchar │    int64     │ varchar  │
├─────────┼──────────────┼──────────┤
│ user_a  │            4 │ Home     │
│ user_a  │            2 │ Settings │
│ user_b  │            2 │ Home     │
└─────────┴──────────────┴──────────┘

Merged partition: u=user_a/d=2023-06-07
2 source files merged files: ['example/_data/u=user_a/d=2023-06-07/103228e7-d363-4ec9-b9b9-fd965a06bdab.parquet', 'example/_data/u=user_a/d=2023-06-07/b72d0af4-6f1f-40f4-b646-a263f199abe6.parquet']
1 new merged file files: ['example/_data/u=user_a/d=2023-06-07/14b31494-569e-4ad5-a72f-5efeac838c0f.parquet']
Merged partition: u=user_b/d=2023-02-11
2 source files merged files: ['example/_data/u=user_b/d=2023-02-11/626680b5-50a5-4934-b640-acd495a35377.parquet', 'example/_data/u=user_b/d=2023-02-11/38fabb73-acc0-489a-9d03-9e10391ca8a7.parquet']
1 new merged file files: ['example/_data/u=user_b/d=2023-02-11/b06dfbf4-ea0e-4e7b-a12d-21e204de54a0.parquet']
Keeping 4 files
3 cleaned log files: ['example/_log/1734871459880_m_dan-mbp.jsonl', 'example/_log/1734871476748_m_dan-mbp.jsonl', 'example/_log/1734871476797_m_dan-mbp.jsonl']
2 deleted log files: ['example/_data/u=user_a/d=2023-06-07/103228e7-d363-4ec9-b9b9-fd965a06bdab.parquet', 'example/_data/u=user_b/d=2023-02-11/626680b5-50a5-4934-b640-acd495a35377.parquet']
2 deleted data files: ['example/_data/u=user_a/d=2023-06-07/103228e7-d363-4ec9-b9b9-fd965a06bdab.parquet', 'example/_data/u=user_b/d=2023-02-11/626680b5-50a5-4934-b640-acd495a35377.parquet']
[2024-12-22 14:44:19 EET]   388B STANDARD 1734871459837_dan-mbp.jsonl
[2024-12-22 14:44:36 EET]   388B STANDARD 1734871476662_dan-mbp.jsonl
[2024-12-22 14:44:36 EET]   672B STANDARD 1734871476815_m_dan-mbp.jsonl
(icedb) taras@b550:~/Downloads/icedb$ python taras/sample.py ;mc ls myminio/testbucket/example/_log/
2 inserted files: ['example/_data/u=user_a/d=2023-06-07/95d918b0-fd16-4a11-b0c0-2e58ba869b4e.parquet', 'example/_data/u=user_b/d=2023-02-11/27fc46b6-f217-49c8-94b7-7743c177e691.parquet']
6 alive files: ['example/_data/u=user_a/d=2023-06-07/103228e7-d363-4ec9-b9b9-fd965a06bdab.parquet', 'example/_data/u=user_b/d=2023-02-11/626680b5-50a5-4934-b640-acd495a35377.parquet', 'example/_data/u=user_a/d=2023-06-07/14b31494-569e-4ad5-a72f-5efeac838c0f.parquet', 'exa
mple/_data/u=user_b/d=2023-02-11/b06dfbf4-ea0e-4e7b-a12d-21e204de54a0.parquet', 'example/_data/u=user_a/d=2023-06-07/95d918b0-fd16-4a11-b0c0-2e58ba869b4e.parquet', 'example/_data/u=user_b/d=2023-02-11/27fc46b6-f217-49c8-94b7-7743c177e691.parquet']
2 tombstoned files: ['example/_data/u=user_a/d=2023-06-07/b72d0af4-6f1f-40f4-b646-a263f199abe6.parquet', 'example/_data/u=user_b/d=2023-02-11/38fabb73-acc0-489a-9d03-9e10391ca8a7.parquet']
Traceback (most recent call last):
  File "/home/taras/Downloads/icedb/taras/sample.py", line 133, in <module>
    print(ddb.sql(query))
          ^^^^^^^^^^^^^^
duckdb.duckdb.HTTPException: HTTP Error: Unable to connect to URL "http://localhost:9900/testbucket/example/_data/u%3Duser_a/d%3D2023-06-07/103228e7-d363-4ec9-b9b9-fd965a06bdab.parquet": 404 (Not Found)
[2024-12-22 14:44:19 EET]   388B STANDARD 1734871459837_dan-mbp.jsonl
[2024-12-22 14:44:36 EET]   388B STANDARD 1734871476662_dan-mbp.jsonl
[2024-12-22 14:44:36 EET]   672B STANDARD 1734871476815_m_dan-mbp.jsonl
[2024-12-22 14:44:46 EET]   388B STANDARD 1734871486500_dan-mbp.jsonl

My modified sample

import duckdb
from icedb.log import S3Client, IceLogIO, FileMarker
from icedb import IceDBv3, CompressionCodec
from datetime import datetime
from time import time

def print_file_stats(description: str, files: list[FileMarker]):
    """Print summary and details of file markers.
    
    Args:
        description: Description of the file group (e.g. "alive", "tombstoned")
        files: List of FileMarker objects to summarize
    """
    paths = list(map(lambda x: x.path, files))
    print(f"{len(paths)} {description} files:", paths)

# S3 configuration dictionary
S3_CONFIG = {
    "s3_region": "us-east-1",
    "s3_endpoint": "http://localhost:9900",
    "s3_access_key_id": "user", 
    "s3_secret_access_key": "password",
    "s3_use_ssl": False,
    "s3_url_style": "path"  # can be 'path' or 'vhost'
}

# Bucket-specific S3 config not used by DuckDB
S3_BUCKET_CONFIG = {
    "bucket": "testbucket",
    "prefix": "example",
}

# create an s3 client to talk to minio
s3c = S3Client(
    s3prefix=S3_BUCKET_CONFIG["prefix"],
    s3bucket=S3_BUCKET_CONFIG["bucket"],
    s3region=S3_CONFIG["s3_region"],
    s3endpoint=S3_CONFIG["s3_endpoint"],
    s3accesskey=S3_CONFIG["s3_access_key_id"],
    s3secretkey=S3_CONFIG["s3_secret_access_key"]
)

example_events = [
    {
        "ts": 1686176939445,
        "event": "page_load",
        "user_id": "user_a",
        "properties": {
            "page_name": "Home"
        }
    }, {
        "ts": 1676126229999,
        "event": "page_load",
        "user_id": "user_b",
        "properties": {
            "page_name": "Home"
        }
    }, {
        "ts": 1686176939666,
        "event": "page_load",
        "user_id": "user_a",
        "properties": {
            "page_name": "Settings"
        }
    }, {
        "ts": 1686176941445,
        "event": "page_load",
        "user_id": "user_a",
        "properties": {
            "page_name": "Home"
        }
    }
]


def part_func(row: dict) -> str:
    """
    Partition by user_id, date
    """
    row_time = datetime.utcfromtimestamp(row['ts'] / 1000)
    part = f"u={row['user_id']}/d={row_time.strftime('%Y-%m-%d')}"
    return part


# Initialize the client
ice = IceDBv3(
    partition_function=part_func,  # Partitions by user_id and date
    sort_order=['event', 'ts'],   # Sort by event, then timestamp of the event within the data part
    # S3 settings from config
    s3_region=S3_CONFIG["s3_region"],
    s3_access_key=S3_CONFIG["s3_access_key_id"],
    s3_secret_key=S3_CONFIG["s3_secret_access_key"],
    s3_endpoint=S3_CONFIG["s3_endpoint"],
    s3_use_path=S3_CONFIG["s3_url_style"] == "path",
    # S3 client instance
    s3_client=s3c,
    # Other settings
    path_safe_hostname="dan-mbp",
    compression_codec=CompressionCodec.ZSTD,  # Use ZSTD for higher compression ratio compared to default SNAPPY
)

# Insert records
inserted = ice.insert(example_events)
print_file_stats("inserted", inserted)

# Read the log state
log = IceLogIO("demo-host")
_, file_markers, _, _ = log.read_at_max_time(s3c, round(time() * 1000))
alive_files = list(filter(lambda x: x.tombstone is None, file_markers))
tombstoned_files = list(filter(lambda x: x.tombstone, file_markers))
print_file_stats("alive", alive_files)
print_file_stats("tombstoned", tombstoned_files)

# Setup duckdb for querying local minio
ddb = duckdb.connect(":memory:")
ddb.execute("install httpfs")
ddb.execute("load httpfs")

# Set DuckDB S3 configuration from the config dictionary
for key, value in S3_CONFIG.items():
    if key == "s3_endpoint":
        # Strip protocol prefix by splitting on :// once
        value = value.split("://", 1)[1]
    ddb.execute(f"SET {key}='{value}'")

# Query alive files
query = ("select user_id, count(*), (properties::JSON)->>'page_name' as page "
         "from read_parquet([{}]) "
         "group by user_id, page "
         "order by count(*) desc").format(
    ', '.join(list(map(lambda x: "'s3://" + ice.data_s3c.s3bucket + "/" + x.path + "'", alive_files)))
)
print(ddb.sql(query))

while True:
    new_log, new_file_marker, partition, merged_file_markers, meta = ice.merge()
    more_to_merge = partition is not None
    if partition:  # if any merge happened
        print(f"Merged partition: {partition}")
        if merged_file_markers:
            print_file_stats("source files merged", merged_file_markers)
        if new_file_marker:
            print_file_stats("new merged file", [new_file_marker])
    else:
        break;
tombstoned = ice.tombstone_cleanup(10_000)
cleaned_logs, deleted_logs, deleted_data = tombstoned
print(f"{len(cleaned_logs)} cleaned log files:", cleaned_logs)
print(f"{len(deleted_logs)} deleted log files:", deleted_logs)
print(f"{len(deleted_data)} deleted data files:", deleted_data)

[tarasglek]

Looks like the problem is that https://github.com/danthegoodman1/icedb/blob/taras/experiment/icedb/log.py#L321 should not be reading from log files older than ones with _m. If I remove them things seem to be correct.

[danthegoodman1]

Looks like the problem is that https://github.com/danthegoodman1/icedb/blob/taras/experiment/icedb/log.py#L321 should not be reading from log files older than ones with _m. If I remove them things seem to be correct.

This would break it, because merging can be concurrent with inserts so it must allow that. It reads the logs in order, so it will detect when log file lines are irrelevant. _m was just a convenience ID suffix to see which were the results of merges.

What's likely happening when you run the example is that log files from previous example runs aren't getting deleted, so on your third run where you get the error, it's trying to read a file that no longer exists because the data file was cleaned, but not the log file.

[danthegoodman1]

Try full wiping minio between your file runs, see if that stop it. If so, then it's just an example cleanup issue, nothing with icedb

[tarasglek]

This is 100% repeatable.
There is no concurrency here, just running insert, insert/merge/tombstone which results in tombstone entries getting dropped.
here we are cleaning up the intermediate _m files during the final merge. But the tombstoned log files that they contain don't get merged into the final _m file.

example/_log/1734875058201_dan-mbp.jsonl
{"v": 1, "sch": 1, "f": 2, "t": 1734875058201}
{"ts": "BIGINT", "event": "VARCHAR", "user_id": "VARCHAR", "properties": "STRUCT(page_name VARCHAR)"}
{"p": "example/_data/u=user_b/d=2023-02-11/e59586dd-bd17-44aa-9308-aff6e652fb27.parquet", "b": 587, "t": 1734875058179}
{"p": "example/_data/u=user_a/d=2023-06-07/da7d294b-f9bd-4c5d-aa4f-34d7ab356237.parquet", "b": 694, "t": 1734875058182}
2 inserted files: ['example/_data/u=user_b/d=2023-02-11/e59586dd-bd17-44aa-9308-aff6e652fb27.parquet', 'example/_data/u=user_a/d=2023-06-07/da7d294b-f9bd-4c5d-aa4f-34d7ab356237.parquet']
read_at_max_time(1734875058204): [{'Key': 'example/_log/1734875055481_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 482000, tzinfo=tzutc()), 'ETag': '"4553212e16c9c2acad1294f60d0e79f1"', 'Size': 388, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875055530_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 530000, tzinfo=tzutc()), 'ETag': '"111f48485cf2b66821afacffb7404829"', 'Size': 49, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058201_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 202000, tzinfo=tzutc()), 'ETag': '"0756955cc6a2047e562c960dcfdd65fb"', 'Size': 388, 'StorageClass': 'STANDARD'}]
read_at_max_time filtered: [{'Key': 'example/_log/1734875055481_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 482000, tzinfo=tzutc()), 'ETag': '"4553212e16c9c2acad1294f60d0e79f1"', 'Size': 388, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875055530_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 530000, tzinfo=tzutc()), 'ETag': '"111f48485cf2b66821afacffb7404829"', 'Size': 49, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058201_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 202000, tzinfo=tzutc()), 'ETag': '"0756955cc6a2047e562c960dcfdd65fb"', 'Size': 388, 'StorageClass': 'STANDARD'}]
4 alive files: ['example/_data/u=user_b/d=2023-02-11/85fe4298-3b6e-4123-b412-2c3dea9d275e.parquet', 'example/_data/u=user_a/d=2023-06-07/71c743ae-fe83-4efb-a05c-a4c69c2e941d.parquet', 'example/_data/u=user_b/d=2023-02-11/e59586dd-bd17-44aa-9308-aff6e652fb27.parquet', 'example/_data/u=user_a/d=2023-06-07/da7d294b-f9bd-4c5d-aa4f-34d7ab356237.parquet']
0 tombstoned files: []
┌─────────┬──────────────┬──────────┐
│ user_id │ count_star() │   page   │
│ varchar │    int64     │ varchar  │
├─────────┼──────────────┼──────────┤
│ user_a  │            4 │ Home     │
│ user_a  │            2 │ Settings │
│ user_b  │            2 │ Home     │
└─────────┴──────────────┴──────────┘

read_at_max_time(1734875058242): [{'Key': 'example/_log/1734875055481_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 482000, tzinfo=tzutc()), 'ETag': '"4553212e16c9c2acad1294f60d0e79f1"', 'Size': 388, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875055530_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 530000, tzinfo=tzutc()), 'ETag': '"111f48485cf2b66821afacffb7404829"', 'Size': 49, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058201_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 202000, tzinfo=tzutc()), 'ETag': '"0756955cc6a2047e562c960dcfdd65fb"', 'Size': 388, 'StorageClass': 'STANDARD'}]
read_at_max_time filtered: [{'Key': 'example/_log/1734875055481_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 482000, tzinfo=tzutc()), 'ETag': '"4553212e16c9c2acad1294f60d0e79f1"', 'Size': 388, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875055530_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 530000, tzinfo=tzutc()), 'ETag': '"111f48485cf2b66821afacffb7404829"', 'Size': 49, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058201_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 202000, tzinfo=tzutc()), 'ETag': '"0756955cc6a2047e562c960dcfdd65fb"', 'Size': 388, 'StorageClass': 'STANDARD'}]
merged_log_files:  ['example/_log/1734875055481_dan-mbp.jsonl', 'example/_log/1734875058201_dan-mbp.jsonl']
new_tombstones [{"p": "example/_log/1734875055481_dan-mbp.jsonl", "t": 1734875058289}, {"p": "example/_log/1734875058201_dan-mbp.jsonl", "t": 1734875058289}]
example/_log/1734875058289_m_dan-mbp.jsonl
{"v": 1, "sch": 1, "f": 4, "t": 1734875058289, "tmb": 2}
{"ts": "BIGINT", "event": "VARCHAR", "user_id": "VARCHAR", "properties": "STRUCT(page_name VARCHAR)"}
{"p": "example/_log/1734875055481_dan-mbp.jsonl", "t": 1734875058289}
{"p": "example/_log/1734875058201_dan-mbp.jsonl", "t": 1734875058289}
{"p": "example/_data/u=user_b/d=2023-02-11/85fe4298-3b6e-4123-b412-2c3dea9d275e.parquet", "b": 587, "t": 1734875055460, "tmb": 1734875058289}
{"p": "example/_data/u=user_a/d=2023-06-07/71c743ae-fe83-4efb-a05c-a4c69c2e941d.parquet", "b": 694, "t": 1734875055463}
{"p": "example/_data/u=user_b/d=2023-02-11/e59586dd-bd17-44aa-9308-aff6e652fb27.parquet", "b": 587, "t": 1734875058179, "tmb": 1734875058289}
{"p": "example/_data/u=user_a/d=2023-06-07/da7d294b-f9bd-4c5d-aa4f-34d7ab356237.parquet", "b": 694, "t": 1734875058182}
{"p": "example/_data/u=user_b/d=2023-02-11/ed966ea9-1919-4a2e-8320-3a706e3daf8c.parquet", "b": 914, "t": 1734875058289}
Merged partition: u=user_b/d=2023-02-11
2 source files merged files: ['example/_data/u=user_b/d=2023-02-11/85fe4298-3b6e-4123-b412-2c3dea9d275e.parquet', 'example/_data/u=user_b/d=2023-02-11/e59586dd-bd17-44aa-9308-aff6e652fb27.parquet']
1 new merged file files: ['example/_data/u=user_b/d=2023-02-11/ed966ea9-1919-4a2e-8320-3a706e3daf8c.parquet']
read_at_max_time(1734875058293): [{'Key': 'example/_log/1734875055481_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 482000, tzinfo=tzutc()), 'ETag': '"4553212e16c9c2acad1294f60d0e79f1"', 'Size': 388, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875055530_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 530000, tzinfo=tzutc()), 'ETag': '"111f48485cf2b66821afacffb7404829"', 'Size': 49, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058201_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 202000, tzinfo=tzutc()), 'ETag': '"0756955cc6a2047e562c960dcfdd65fb"', 'Size': 388, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058289_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 289000, tzinfo=tzutc()), 'ETag': '"3341f390fcb36566061c0055a0af92f4"', 'Size': 942, 'StorageClass': 'STANDARD'}]
read_at_max_time filtered: [{'Key': 'example/_log/1734875055481_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 482000, tzinfo=tzutc()), 'ETag': '"4553212e16c9c2acad1294f60d0e79f1"', 'Size': 388, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875055530_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 530000, tzinfo=tzutc()), 'ETag': '"111f48485cf2b66821afacffb7404829"', 'Size': 49, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058201_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 202000, tzinfo=tzutc()), 'ETag': '"0756955cc6a2047e562c960dcfdd65fb"', 'Size': 388, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058289_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 289000, tzinfo=tzutc()), 'ETag': '"3341f390fcb36566061c0055a0af92f4"', 'Size': 942, 'StorageClass': 'STANDARD'}]
merged_log_files:  ['example/_log/1734875058289_m_dan-mbp.jsonl', 'example/_log/1734875058289_m_dan-mbp.jsonl']
new_tombstones [{"p": "example/_log/1734875058289_m_dan-mbp.jsonl", "t": 1734875058338}, {"p": "example/_log/1734875058289_m_dan-mbp.jsonl", "t": 1734875058338}]
example/_log/1734875058338_m_dan-mbp.jsonl
{"v": 1, "sch": 1, "f": 6, "t": 1734875058338, "tmb": 2}
{"ts": "BIGINT", "event": "VARCHAR", "user_id": "VARCHAR", "properties": "STRUCT(page_name VARCHAR)"}
{"p": "example/_log/1734875055481_dan-mbp.jsonl", "t": 1734875058289}
{"p": "example/_log/1734875058201_dan-mbp.jsonl", "t": 1734875058289}
{"p": "example/_log/1734875058289_m_dan-mbp.jsonl", "t": 1734875058338}
{"p": "example/_log/1734875058289_m_dan-mbp.jsonl", "t": 1734875058338}
{"p": "example/_data/u=user_b/d=2023-02-11/85fe4298-3b6e-4123-b412-2c3dea9d275e.parquet", "b": 587, "t": 1734875055460, "tmb": 1734875058289}
{"p": "example/_data/u=user_a/d=2023-06-07/71c743ae-fe83-4efb-a05c-a4c69c2e941d.parquet", "b": 694, "t": 1734875055463, "tmb": 1734875058338}
{"p": "example/_data/u=user_b/d=2023-02-11/e59586dd-bd17-44aa-9308-aff6e652fb27.parquet", "b": 587, "t": 1734875058179, "tmb": 1734875058289}
{"p": "example/_data/u=user_a/d=2023-06-07/da7d294b-f9bd-4c5d-aa4f-34d7ab356237.parquet", "b": 694, "t": 1734875058182, "tmb": 1734875058338}
{"p": "example/_data/u=user_b/d=2023-02-11/ed966ea9-1919-4a2e-8320-3a706e3daf8c.parquet", "b": 914, "t": 1734875058289}
{"p": "example/_data/u=user_a/d=2023-06-07/3dc0078b-22d0-49ac-8bcb-600a7c88739f.parquet", "b": 957, "t": 1734875058338}
Merged partition: u=user_a/d=2023-06-07
2 source files merged files: ['example/_data/u=user_a/d=2023-06-07/71c743ae-fe83-4efb-a05c-a4c69c2e941d.parquet', 'example/_data/u=user_a/d=2023-06-07/da7d294b-f9bd-4c5d-aa4f-34d7ab356237.parquet']
1 new merged file files: ['example/_data/u=user_a/d=2023-06-07/3dc0078b-22d0-49ac-8bcb-600a7c88739f.parquet']
read_at_max_time(1734875058342): [{'Key': 'example/_log/1734875055481_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 482000, tzinfo=tzutc()), 'ETag': '"4553212e16c9c2acad1294f60d0e79f1"', 'Size': 388, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875055530_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 530000, tzinfo=tzutc()), 'ETag': '"111f48485cf2b66821afacffb7404829"', 'Size': 49, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058201_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 202000, tzinfo=tzutc()), 'ETag': '"0756955cc6a2047e562c960dcfdd65fb"', 'Size': 388, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058289_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 289000, tzinfo=tzutc()), 'ETag': '"3341f390fcb36566061c0055a0af92f4"', 'Size': 942, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058338_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 338000, tzinfo=tzutc()), 'ETag': '"5d6eae01d450a09bd02390ca54497957"', 'Size': 1250, 'StorageClass': 'STANDARD'}]
read_at_max_time filtered: [{'Key': 'example/_log/1734875055481_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 482000, tzinfo=tzutc()), 'ETag': '"4553212e16c9c2acad1294f60d0e79f1"', 'Size': 388, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875055530_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 15, 530000, tzinfo=tzutc()), 'ETag': '"111f48485cf2b66821afacffb7404829"', 'Size': 49, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058201_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 202000, tzinfo=tzutc()), 'ETag': '"0756955cc6a2047e562c960dcfdd65fb"', 'Size': 388, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058289_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 289000, tzinfo=tzutc()), 'ETag': '"3341f390fcb36566061c0055a0af92f4"', 'Size': 942, 'StorageClass': 'STANDARD'}, {'Key': 'example/_log/1734875058338_m_dan-mbp.jsonl', 'LastModified': datetime.datetime(2024, 12, 22, 13, 44, 18, 338000, tzinfo=tzutc()), 'ETag': '"5d6eae01d450a09bd02390ca54497957"', 'Size': 1250, 'StorageClass': 'STANDARD'}]
example/_log/1734875058353_m_dan-mbp.jsonl
{"v": 1, "sch": 1, "f": 2, "t": 1734875058353}
{"ts": "BIGINT", "event": "VARCHAR", "user_id": "VARCHAR", "properties": "STRUCT(page_name VARCHAR)"}
{"p": "example/_data/u=user_b/d=2023-02-11/85fe4298-3b6e-4123-b412-2c3dea9d275e.parquet", "b": 587, "t": 1734875055460, "tmb": 1734875058289}
{"p": "example/_data/u=user_a/d=2023-06-07/71c743ae-fe83-4efb-a05c-a4c69c2e941d.parquet", "b": 694, "t": 1734875055463, "tmb": 1734875058338}
{"p": "example/_data/u=user_b/d=2023-02-11/e59586dd-bd17-44aa-9308-aff6e652fb27.parquet", "b": 587, "t": 1734875058179, "tmb": 1734875058289}
{"p": "example/_data/u=user_a/d=2023-06-07/da7d294b-f9bd-4c5d-aa4f-34d7ab356237.parquet", "b": 694, "t": 1734875058182, "tmb": 1734875058338}
{"p": "example/_data/u=user_b/d=2023-02-11/ed966ea9-1919-4a2e-8320-3a706e3daf8c.parquet", "b": 914, "t": 1734875058289}
{"p": "example/_data/u=user_a/d=2023-06-07/3dc0078b-22d0-49ac-8bcb-600a7c88739f.parquet", "b": 957, "t": 1734875058338}
Keeping 6 files
3 cleaned log files: ['example/_log/1734875055530_m_dan-mbp.jsonl', 'example/_log/1734875058289_m_dan-mbp.jsonl', 'example/_log/1734875058338_m_dan-mbp.jsonl']
0 deleted log files: []
0 deleted data files: []

this results in log files:

[2024-12-22 15:43:17 EET]   388B STANDARD 1734874997398_dan-mbp.jsonl
[2024-12-22 15:43:24 EET]   388B STANDARD 1734875004316_dan-mbp.jsonl
[2024-12-22 15:43:24 EET]   956B STANDARD 1734875004470_m_dan-mbp.jsonl

[tarasglek]

Eg..the intermediate merge files above contain:

{"p": "example/_log/1734875055481_dan-mbp.jsonl", "t": 1734875058289}
{"p": "example/_log/1734875058201_dan-mbp.jsonl", "t": 1734875058289}
{"p": "example/_log/1734875058289_m_dan-mbp.jsonl", "t": 1734875058338}
{"p": "example/_log/1734875058289_m_dan-mbp.jsonl", "t": 1734875058338}
{"p": "example/_data/u=user_b/d=2023-02-11/85fe4298-3b6e-4123-b412-2c3dea9d275e.parquet", "b": 587, "t": 1734875055460, "tmb": 1734875058289}
{"p": "example/_data/u=user_a/d=2023-06-07/71c743ae-fe83-4efb-a05c-a4c69c2e941d.parquet", "b": 694, "t": 1734875055463, "tmb": 1734875058338}

and the logic in https://github.com/tarasglek/icedb/blob/taras/experiment/icedb/icedb.py#L368 neither flags these log files for deletion or carries over these tombstones to 1734875004470_m_dan-mbp.jsonl

I do not understand why logs being added as tombstones produced in merge func https://github.com/tarasglek/icedb/blob/taras/experiment/icedb/icedb.py#L309 don't have tmb flag that cleanup func expects https://github.com/tarasglek/icedb/blob/taras/experiment/icedb/icedb.py#L368. Is there some reason for that or is that accidental?

[danthegoodman1]

The files you're linking to are in a branch that doesn't exist.

I just ran the simple-full.py script 12 times in a row after removing the delete_all_s3 and it seems to work as expected:

============= tombstone cleaning =============
Keeping 4 files
cleaned 2 log files and deleted 11 log files and deleted 5 data files
============= running query =============
┌─────────┬──────────────┬──────────┐
│ user_id │ count_star() │   page   │
│ varchar │    int64     │ varchar  │
├─────────┼──────────────┼──────────┤
│ user_a  │           48 │ Home     │
│ user_a  │           24 │ Settings │
│ user_b  │           24 │ Home     │
└─────────┴──────────────┴──────────┘

Something seems strange with your code, why is the deleted log files printing parquet files?

2 deleted log files: ['example/_data/u=user_a/d=2023-06-07/103228e7-d363-4ec9-b9b9-fd965a06bdab.parquet', 'example/_data/u=user_b/d=2023-02-11/626680b5-50a5-4934-b640-acd495a35377.parquet']

[tarasglek]

The files you're linking to are in a branch that doesn't exist.

I updated links. Sorry.

My understanding is that the .merge code runs 3 times. 3rd merge results in missing log tombstones (or more likely missing delete operations on those log files);

I just ran the simple-full.py script 12 times in a row after removing the delete_all_s3 and it seems to work as expected:

============= tombstone cleaning =============
Keeping 4 files
cleaned 2 log files and deleted 11 log files and deleted 5 data files
============= running query =============
┌─────────┬──────────────┬──────────┐
│ user_id │ count_star() │   page   │
│ varchar │    int64     │ varchar  │
├─────────┼──────────────┼──────────┤
│ user_a  │           48 │ Home     │
│ user_a  │           24 │ Settings │
│ user_b  │           24 │ Home     │
└─────────┴──────────────┴──────────┘

I run docker compose down -v to reset minio, I guess it has same effect.

I run docker compose down -v; docker compose up -d, then run my script 3 times to get the error.

My script is not the same as your simple-full. I attached in above, here is current version: https://github.com/tarasglek/icedb/blob/taras/experiment/sample.py

python sample.py ;mc ls myminio/testbucket/example/_log/

The main difference is that it calls .merge until there is no more partitions left to merge.

Something seems strange with your code, why is the deleted log files printing parquet files?

2 deleted log files: ['example/_data/u=user_a/d=2023-06-07/103228e7-d363-4ec9-b9b9-fd965a06bdab.parquet', 'example/_data/u=user_b/d=2023-02-11/626680b5-50a5-4934-b640-acd495a35377.parquet']

There is a typo here:
https://github.com/danthegoodman1/icedb/blob/main/icedb/icedb.py#L397

[danthegoodman1]

I had to run your script (the one posted above) 6-8 times before the error occurred.

Good catch on that error, thanks. Let me look at this for a bit to make sure I fully understand.

[danthegoodman1]

I do not understand why logs being added as tombstones produced in merge func https://github.com/tarasglek/icedb/blob/taras/experiment/icedb/icedb.py#L309 don't have tmb flag that cleanup func expects https://github.com/tarasglek/icedb/blob/taras/experiment/icedb/icedb.py#L368. Is there some reason for that or is that accidental?

You're mixing log tombstones and data file tombstones. LogTombstone doesn't need an explicit tombstone timestamp because it's existence is a tombstone. The log format indicates at what offset these begin, so we know a certain range of lines in the log file are all log tombstones.

[danthegoodman1]

Frankly I haven't touched this codebase in a while, so it's not all loaded into my head to be able to quickly fix it, so I'd encourage you to submit a PR :D

(make sure all the examples still work, as well as the tests in the root folder, which you'll have to pip uninstall icedb to make sure you're running with the local version, not the pip installed version).

12-22-2024-10-40-39_files_list.zip
Here's a minimal set of logs that recreates the error, a few things I'm noticing:

The files from the ...938_...jsonl log file:

{"v": 1, "sch": 1, "f": 2, "t": 1734880893938}
{"ts": "BIGINT", "event": "VARCHAR", "user_id": "VARCHAR", "properties": "STRUCT(page_name VARCHAR)"}
{"p": "example/_data/u=user_a/d=2023-06-07/e2b1d1e2-8de6-4ce2-83fb-70dfffa7d506.parquet", "b": 694, "t": 1734880893923}
{"p": "example/_data/u=user_b/d=2023-02-11/150e78a9-076a-4523-b64f-eddd67f48cd2.parquet", "b": 587, "t": 1734880893934}

Seem to be merged with the ...244_...jsonl log file:

{"v": 1, "sch": 1, "f": 2, "t": 1734881096244}
{"ts": "BIGINT", "event": "VARCHAR", "user_id": "VARCHAR", "properties": "STRUCT(page_name VARCHAR)"}
{"p": "example/_data/u=user_b/d=2023-02-11/5052e29e-2d01-4b90-b8aa-5b95cfa578e9.parquet", "b": 587, "t": 1734881096222}
{"p": "example/_data/u=user_a/d=2023-06-07/cf88bb24-2f9a-4ef7-ab74-80ca22d8e1e9.parquet", "b": 694, "t": 1734881096237}

info the ...428_m_...jsonl, but I don't see the tombstones for the files from the first log (only the second):

{"v": 1, "sch": 1, "f": 2, "t": 1734881096428}
{"ts": "BIGINT", "event": "VARCHAR", "user_id": "VARCHAR", "properties": "STRUCT(page_name VARCHAR)"}
{"p": "example/_data/u=user_b/d=2023-02-11/5052e29e-2d01-4b90-b8aa-5b95cfa578e9.parquet", "b": 587, "t": 1734881096222, "tmb": 1734881096405}
{"p": "example/_data/u=user_a/d=2023-06-07/cf88bb24-2f9a-4ef7-ab74-80ca22d8e1e9.parquet", "b": 694, "t": 1734881096237, "tmb": 1734881096355}
{"p": "example/_data/u=user_a/d=2023-06-07/5566dbe9-56fd-4e9e-813f-456862d49156.parquet", "b": 957, "t": 1734881096355}
{"p": "example/_data/u=user_b/d=2023-02-11/a2fcc74c-1951-437b-82d7-e4351cd48a3f.parquet", "b": 914, "t": 1734881096405}

So e2b1d1e2-8de6-4ce2-83fb-70dfffa7d506.parquet is attempted to be read from the first log file, which should have been merged into 5566dbe9-56fd-4e9e-813f-456862d49156.parquet. However that first log file shouldn't exist. If you remove the tombstone part of your code, you can see that log files and properly marked as tombstones

I suspect there is an issue with either the tombstone cleanup not deleting merged log files, or merged log files not getting marked as merged. One thing I'm noticing there are duplicate tombstone log markers in there ifyou merge them up multiple times, but that doesn't affect anything.

In another run with extra logging, the error of:

Came from the log file 1734883028124_dan-mbp which I see in logs:

That log file seems like it should have been cleaned up.

I would suggest adding logs to all the points where file are deleted and merged. My guess is that tombstone cleaning is not deleting the log (maybe we silently exit a loop if we hit a duplicate tombstone log entry like i'm seeing here:

When I disable tombstone cleaning, everything is working just fine, so I have a feeling for some reason tombstone cleaning is aborting deleting files, and since the tombstone markers are cleaned from the log, but the files are not deleted, they're considered alive and valid (despite the data files already being deleted).

[danthegoodman1]

There is a typo here:
https://github.com/danthegoodman1/icedb/blob/main/icedb/icedb.py#L397

Watch out for this, and any other log mistakes I might have made :P (def include them in the PR!)