24057299873

Committed 07 Apr 2026 12:01AM UTC coverage: 52.076% (+0.8%) from 51.228%

Build # 24057299873

Build Type

push

github

Committed by

rayandrew

Commit Message

feat(rocksdb): migrate SQLite indexing to RocksDB

Replace SQLite-backed indexing and provenance storage with RocksDB-backed stores.

  Key changes:
  - add RocksDB async/database/db-manager/filesystem/key-codec layers
  - migrate index and provenance databases from SQLite to RocksDB
  - update index builder, trace reader, reorganize, view, stats, and comparator paths for
  RocksDB
  - harden transaction atomicity and rollback behavior with TransactionScope
  - add iterator status checking for prefix scans
  - harden gzip/tar indexer cache state and metadata handling
  - capture executor context in RocksDB awaitables
  - clean up failed RocksDB open paths and manager lifecycle behavior
  - vendor CPM 0.42.1 and update CI/build integration
  - refresh docs, Python bindings, and C++/Python test coverage for the new backend

  Validation:
  - full test suite passed
  - Ubuntu 22.04 Docker run passed
  - focused RocksDB/indexer regression tests passed.

Coverage Stats

24097 of 59624 branches covered (40.41%)

Branch coverage included in aggregate %.

2516 of 3144 new or added lines in 75 files covered. (80.03%)

72 existing lines in 15 files now uncovered.

20858 of 26701 relevant lines covered (78.12%)

14113.43 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

62.43

/src/dftracer/utils/python/indexer.cpp

#include <dftracer/utils/core/runtime.h>
#include <dftracer/utils/python/indexer.h>
#include <dftracer/utils/python/indexer_checkpoint.h>
#include <dftracer/utils/python/runtime.h>
#include <dftracer/utils/utilities/composites/dft/internal/utils.h>
#include <dftracer/utils/utilities/indexer/index_builder_utility.h>
#include <dftracer/utils/utilities/indexer/index_database.h>
#include <dftracer/utils/utilities/indexer/internal/helpers.h>
#include <structmember.h>

#include <cstring>

static void Indexer_dealloc(IndexerObject *self) {
    if (self->handle) {
        // The Python wrapper owns only the native indexer handle. The
        // underlying RocksDB instance remains manager-owned and may continue to
        // live process-wide for the same .dftindex path.
        dft_indexer_destroy(self->handle);
        self->handle = NULL;
    }
    Py_XDECREF(self->gz_path);
    Py_XDECREF(self->index_path);
    Py_XDECREF(self->runtime_obj);
    Py_TYPE(self)->tp_free((PyObject *)self);
}

static void Indexer_release_handle(IndexerObject *self) {
    if (self->handle) {
        // Releasing the handle drops this wrapper's native indexer state only.
        // Shared RocksDB lifetime is managed separately by RocksDBManager.
        dft_indexer_destroy(self->handle);
        self->handle = NULL;
    }
}

static PyObject *Indexer_new(PyTypeObject *type, PyObject *args,
                             PyObject *kwds) {
    IndexerObject *self;
    self = (IndexerObject *)type->tp_alloc(type, 0);
    if (self != NULL) {
        self->handle = NULL;
        self->gz_path = NULL;
        self->index_path = NULL;
        self->checkpoint_size = 0;
        self->build_bloom = 0;
        self->build_manifest = 0;
        self->index_threshold =
            dftracer::utils::constants::indexer::DEFAULT_INDEX_SIZE_THRESHOLD;
        self->runtime_obj = NULL;
    }
    return (PyObject *)self;
}

static int Indexer_init(IndexerObject *self, PyObject *args, PyObject *kwds) {
    static const char *kwlist[] = {
        "gz_path",         "index_path",  "checkpoint_size",
        "force_rebuild",   "build_bloom", "build_manifest",
        "index_threshold", "runtime",     NULL};
    const char *gz_path;
    const char *index_path = NULL;
    std::uint64_t checkpoint_size =
        dftracer::utils::constants::indexer::DEFAULT_CHECKPOINT_SIZE;
    int force_rebuild = 0;
    int build_bloom = 0;
    int build_manifest = 0;
    std::uint64_t index_threshold =
        dftracer::utils::constants::indexer::DEFAULT_INDEX_SIZE_THRESHOLD;
    PyObject *runtime_arg = NULL;

    if (!PyArg_ParseTupleAndKeywords(
            args, kwds, "s|snpppnO", (char **)kwlist, &gz_path, &index_path,
            &checkpoint_size, &force_rebuild, &build_bloom, &build_manifest,
            &index_threshold, &runtime_arg)) {
        return -1;
    }

    if (runtime_arg && runtime_arg != Py_None) {
        if (PyObject_TypeCheck(runtime_arg, &RuntimeType)) {
            Py_INCREF(runtime_arg);
            self->runtime_obj = runtime_arg;
        } else {
            PyObject *native = PyObject_GetAttrString(runtime_arg, "_native");
            if (native && PyObject_TypeCheck(native, &RuntimeType)) {
                self->runtime_obj = native;
            } else {
                Py_XDECREF(native);
                PyErr_SetString(PyExc_TypeError,
                                "runtime must be a Runtime instance or None");
                return -1;
            }
        }
    }

    self->gz_path = PyUnicode_FromString(gz_path);
    if (!self->gz_path) {
        return -1;
    }

    if (index_path) {
        self->index_path = PyUnicode_FromString(index_path);
    } else {
        const std::string index_path = dftracer::utils::utilities::composites::
            dft::internal::determine_index_path(gz_path, "");
        self->index_path = PyUnicode_FromString(index_path.c_str());
    }

    if (!self->index_path) {
        Py_DECREF(self->gz_path);
        return -1;
    }

    self->checkpoint_size = checkpoint_size;
    self->build_bloom = build_bloom;
    self->build_manifest = build_manifest;
    self->index_threshold = index_threshold;

    const char *index_path_str = PyUnicode_AsUTF8(self->index_path);
    if (!index_path_str) {
        return -1;
    }

    self->handle = dft_indexer_create(gz_path, index_path_str, checkpoint_size,
                                      force_rebuild);
    if (!self->handle) {
        PyErr_SetString(PyExc_RuntimeError, "Failed to create indexer");
        return -1;
    }

    return 0;
}

static dftracer::utils::Runtime *get_indexer_runtime(IndexerObject *self) {
    if (self->runtime_obj) {
        return ((RuntimeObject *)self->runtime_obj)->runtime.get();
    }
    return get_default_runtime();
}

static PyObject *Indexer_build(IndexerObject *self,
                               PyObject *Py_UNUSED(ignored)) {
    if (!self->handle) {
        PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");
        return NULL;
    }

    using namespace dftracer::utils;
    using namespace dftracer::utils::utilities::indexer;

    const char *gz = PyUnicode_AsUTF8(self->gz_path);
    const char *idx = PyUnicode_AsUTF8(self->index_path);
    if (!gz || !idx) {
        return NULL;
    }

    auto config = IndexBuildConfig::for_file(gz)
                      .with_checkpoint_size(
                          static_cast<std::size_t>(self->checkpoint_size))
                      .with_bloom(self->build_bloom != 0)
                      .with_manifest(self->build_manifest != 0)
                      .with_index_threshold(
                          static_cast<std::size_t>(self->index_threshold));

    std::string idx_str(idx);
    auto pos = idx_str.find_last_of('/');
    if (pos != std::string::npos) {
        config.with_index_dir(idx_str.substr(0, pos));
    }

    Runtime *rt = get_indexer_runtime(self);
    IndexBuildResult build_result;

    try {
        auto build_coro =
            [](IndexBuildConfig cfg) -> coro::CoroTask<IndexBuildResult> {
            IndexBuilderUtility builder;
            co_return co_await builder.process(cfg);
        };

        Py_BEGIN_ALLOW_THREADS auto handle =
            rt->submit(build_coro(config), "indexer-build");
        build_result = handle.get();
        Py_END_ALLOW_THREADS
    } catch (const std::exception &e) {
        PyErr_SetString(PyExc_RuntimeError, e.what());
        return NULL;
    }

    if (!build_result.success) {
        PyErr_SetString(PyExc_RuntimeError, build_result.error_message.c_str());
        return NULL;
    }

    Py_RETURN_NONE;
}

static PyObject *Indexer_need_rebuild(IndexerObject *self,
                                      PyObject *Py_UNUSED(ignored)) {
    if (!self->handle) {
        PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");
        return NULL;
    }

    int result = dft_indexer_need_rebuild(self->handle);
    return PyBool_FromLong(result);
}

static PyObject *Indexer_exists(IndexerObject *self,
                                PyObject *Py_UNUSED(ignored)) {
    if (!self->handle) {
        PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");
        return NULL;
    }

    int result = dft_indexer_exists(self->handle);
    return PyBool_FromLong(result);
}

static PyObject *Indexer_get_max_bytes(IndexerObject *self,
                                       PyObject *Py_UNUSED(ignored)) {
    if (!self->handle) {
        PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");
        return NULL;
    }

    uint64_t result = dft_indexer_get_max_bytes(self->handle);
    return PyLong_FromUnsignedLongLong(result);
}

static PyObject *Indexer_get_num_lines(IndexerObject *self,
                                       PyObject *Py_UNUSED(ignored)) {
    if (!self->handle) {
        PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");
        return NULL;
    }

    uint64_t result = dft_indexer_get_num_lines(self->handle);
    return PyLong_FromUnsignedLongLong(result);
}

static PyObject *Indexer_find_checkpoint(IndexerObject *self, PyObject *args) {
    if (!self->handle) {
        PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");
        return NULL;
    }

    std::size_t target_offset;
    if (!PyArg_ParseTuple(args, "n", &target_offset)) {
        return NULL;
    }

    dft_indexer_checkpoint_t checkpoint;
    int found =
        dft_indexer_find_checkpoint(self->handle, target_offset, &checkpoint);

    if (!found) {
        Py_RETURN_NONE;
    }

    // Create IndexerCheckpoint object
    IndexerCheckpointObject *cp_obj =
        (IndexerCheckpointObject *)IndexerCheckpoint_new(&IndexerCheckpointType,
                                                         NULL, NULL);
    if (!cp_obj) {
        return NULL;
    }

    cp_obj->checkpoint = checkpoint;
    return (PyObject *)cp_obj;
}

static PyObject *Indexer_get_checkpoints(IndexerObject *self,
                                         PyObject *Py_UNUSED(ignored)) {
    if (!self->handle) {
        PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");
        return NULL;
    }

    dft_indexer_checkpoint_t *checkpoints = NULL;
    std::size_t count = 0;

    int result =
        dft_indexer_get_checkpoints(self->handle, &checkpoints, &count);
    if (result != 0 || !checkpoints) {
        dft_indexer_free_checkpoints(checkpoints, count);
        PyObject *list = PyList_New(0);
        return list;
    }

    PyObject *list = PyList_New(count);
    if (!list) {
        dft_indexer_free_checkpoints(checkpoints, count);
        return NULL;
    }

    for (std::size_t i = 0; i < count; i++) {
        IndexerCheckpointObject *cp_obj =
            (IndexerCheckpointObject *)IndexerCheckpoint_new(
                &IndexerCheckpointType, NULL, NULL);
        if (!cp_obj) {
            Py_DECREF(list);
            dft_indexer_free_checkpoints(checkpoints, count);
            return NULL;
        }
        cp_obj->checkpoint = checkpoints[i];
        PyList_SetItem(list, i, (PyObject *)cp_obj);
    }

    dft_indexer_free_checkpoints(checkpoints, count);
    return list;
}

static PyObject *Indexer_has_bloom(IndexerObject *self, void *closure) {
    const char *idx = PyUnicode_AsUTF8(self->index_path);
    const char *gz = PyUnicode_AsUTF8(self->gz_path);
    if (!idx || !gz) {
        Py_RETURN_FALSE;
    }
    try {
        using namespace dftracer::utils::utilities::indexer;
        using namespace dftracer::utils::utilities::indexer::internal;
        IndexDatabase db(
            idx, dftracer::utils::rocksdb::RocksDatabase::OpenMode::ReadOnly);
        std::string logical = get_logical_path(gz);
        int fid = db.get_file_info_id(logical);
        if (fid >= 0 && db.has_bloom_data(fid)) {
            Py_RETURN_TRUE;
        }
    } catch (...) {
    }
    Py_RETURN_FALSE;
}

static PyObject *Indexer_has_manifest(IndexerObject *self, void *closure) {
    const char *idx = PyUnicode_AsUTF8(self->index_path);
    const char *gz = PyUnicode_AsUTF8(self->gz_path);
    if (!idx || !gz) {
        Py_RETURN_FALSE;
    }
    try {
        using namespace dftracer::utils::utilities::indexer;
        using namespace dftracer::utils::utilities::indexer::internal;
        IndexDatabase db(
            idx, dftracer::utils::rocksdb::RocksDatabase::OpenMode::ReadOnly);
        std::string logical = get_logical_path(gz);
        int fid = db.get_file_info_id(logical);
        if (fid >= 0 && db.has_manifest_data(fid)) {
            Py_RETURN_TRUE;
        }
    } catch (...) {
    }
    Py_RETURN_FALSE;
}

static PyObject *Indexer_gz_path(IndexerObject *self, void *closure) {
    Py_INCREF(self->gz_path);
    return self->gz_path;
}

static PyObject *Indexer_index_path(IndexerObject *self, void *closure) {
    Py_INCREF(self->index_path);
    return self->index_path;
}

static PyObject *Indexer_checkpoint_size(IndexerObject *self, void *closure) {
    return PyLong_FromUnsignedLongLong(self->checkpoint_size);
}

static PyObject *Indexer_enter(IndexerObject *self,
                               PyObject *Py_UNUSED(ignored)) {
    Py_INCREF(self);
    return (PyObject *)self;
}

static PyObject *Indexer_close(IndexerObject *self,
                               PyObject *Py_UNUSED(ignored)) {
    Indexer_release_handle(self);
    Py_RETURN_NONE;
}

static PyObject *Indexer_exit(IndexerObject *self, PyObject *args) {
    Indexer_release_handle(self);
    Py_RETURN_NONE;
}

static PyMethodDef Indexer_methods[] = {
    {"build", (PyCFunction)Indexer_build, METH_NOARGS,
     "build()\n"
     "--\n"
     "\n"
     "Build or rebuild the index.\n"},
    {"need_rebuild", (PyCFunction)Indexer_need_rebuild, METH_NOARGS,
     "Check if a rebuild is needed."},
    {"exists", (PyCFunction)Indexer_exists, METH_NOARGS,
     "Check if the .dftindex store exists."},
    {"get_max_bytes", (PyCFunction)Indexer_get_max_bytes, METH_NOARGS,
     "Get the maximum uncompressed bytes in the indexed file."},
    {"get_num_lines", (PyCFunction)Indexer_get_num_lines, METH_NOARGS,
     "Get the total number of lines in the indexed file."},
    {"find_checkpoint", (PyCFunction)Indexer_find_checkpoint, METH_VARARGS,
     "Find the best checkpoint for a given uncompressed offset.\n"
     "\n"
     "Args:\n"
     "    offset (int): Uncompressed byte offset.\n"},
    {"get_checkpoints", (PyCFunction)Indexer_get_checkpoints, METH_NOARGS,
     "Get all checkpoints for this file as a list."},
    {"close", (PyCFunction)Indexer_close, METH_NOARGS,
     "Release this Python wrapper's native indexer handle.\n"
     "\n"
     "The shared RocksDB instance for the same .dftindex path remains managed\n"
     "by the native RocksDBManager cache."},
    {"__enter__", (PyCFunction)Indexer_enter, METH_NOARGS,
     "Enter the runtime context for the with statement."},
    {"__exit__", (PyCFunction)Indexer_exit, METH_VARARGS,
     "Release this Python wrapper on context exit.\n"
     "\n"
     "This does not force-close the shared RocksDB instance for the same\n"
     ".dftindex path."},
    {NULL} /* Sentinel */
};

static PyGetSetDef Indexer_getsetters[] = {
    {"gz_path", (getter)Indexer_gz_path, NULL, "Path to the gzip file", NULL},
    {"index_path", (getter)Indexer_index_path, NULL,
     "Path to the .dftindex store", NULL},
    {"checkpoint_size", (getter)Indexer_checkpoint_size, NULL,
     "Checkpoint size in bytes", NULL},
    {"has_bloom", (getter)Indexer_has_bloom, NULL,
     "Whether bloom data exists in index", NULL},
    {"has_manifest", (getter)Indexer_has_manifest, NULL,
     "Whether manifest data exists in index", NULL},
    {NULL} /* Sentinel */
};

PyTypeObject IndexerType = {
    PyVarObject_HEAD_INIT(NULL, 0) "indexer.Indexer", /* tp_name */
    sizeof(IndexerObject),                            /* tp_basicsize */
    0,                                                /* tp_itemsize */
    (destructor)Indexer_dealloc,                      /* tp_dealloc */
    0,                                                /* tp_vectorcall_offset */
    0,                                                /* tp_getattr */
    0,                                                /* tp_setattr */
    0,                                                /* tp_as_async */
    0,                                                /* tp_repr */
    0,                                                /* tp_as_number */
    0,                                                /* tp_as_sequence */
    0,                                                /* tp_as_mapping */
    0,                                                /* tp_hash */
    0,                                                /* tp_call */
    0,                                                /* tp_str */
    0,                                                /* tp_getattro */
    0,                                                /* tp_setattro */
    0,                                                /* tp_as_buffer */
    Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE,         /* tp_flags */
    "Indexer(gz_path: str, index_path: str | None = None,\n"
    "       checkpoint_size: int = 1048576,\n"
    "       force_rebuild: bool = False, build_bloom: bool = False,\n"
    "       build_manifest: bool = False,\n"
    "       index_threshold: int = 1048576,\n"
    "       runtime: Runtime | None = None)\n"
    "--\n"
    "\n"
    "Indexer for creating and managing gzip trace index stores.\n"
    "\n"
    "Args:\n"
    "    gz_path (str): Path to the gzip trace file.\n"
    "    index_path (str or None): Path to the .dftindex store. If None,\n"
    "        uses the root-local \".dftindex\" next to gz_path.\n"
    "    checkpoint_size (int): Checkpoint size in bytes for index\n"
    "        building (default 1 MB).\n"
    "    force_rebuild (bool): If True, rebuild the index even if it\n"
    "        exists.\n"
    "    build_bloom (bool): If True, build bloom filter data in the\n"
    "        store.\n"
    "    build_manifest (bool): If True, build manifest data in the\n"
    "        store.\n"
    "    index_threshold (int): Skip indexing for files smaller than\n"
    "        this (default 1 MB).\n"
    "    runtime (Runtime or None): Runtime instance for thread pool\n"
    "        control. If None, uses the default global Runtime.\n", /* tp_doc */
    0,                      /* tp_traverse */
    0,                      /* tp_clear */
    0,                      /* tp_richcompare */
    0,                      /* tp_weaklistoffset */
    0,                      /* tp_iter */
    0,                      /* tp_iternext */
    Indexer_methods,        /* tp_methods */
    0,                      /* tp_members */
    Indexer_getsetters,     /* tp_getset */
    0,                      /* tp_base */
    0,                      /* tp_dict */
    0,                      /* tp_descr_get */
    0,                      /* tp_descr_set */
    0,                      /* tp_dictoffset */
    (initproc)Indexer_init, /* tp_init */
    0,                      /* tp_alloc */
    Indexer_new,            /* tp_new */
};

int init_indexer(PyObject *m) {
    if (PyType_Ready(&IndexerType) < 0) return -1;

    Py_INCREF(&IndexerType);
    if (PyModule_AddObject(m, "Indexer", (PyObject *)&IndexerType) < 0) {
        Py_DECREF(&IndexerType);
        Py_DECREF(m);
        return -1;
    }

    return 0;
}

1	#include <dftracer/utils/core/runtime.h>
2	#include <dftracer/utils/python/indexer.h>
3	#include <dftracer/utils/python/indexer_checkpoint.h>
4	#include <dftracer/utils/python/runtime.h>
5	#include <dftracer/utils/utilities/composites/dft/internal/utils.h>
6	#include <dftracer/utils/utilities/indexer/index_builder_utility.h>
7	#include <dftracer/utils/utilities/indexer/index_database.h>
8	#include <dftracer/utils/utilities/indexer/internal/helpers.h>
9	#include <structmember.h>
10
11	#include <cstring>
12
13	static void Indexer_dealloc(IndexerObject *self) {	130✔
14	if (self->handle) {	130✔
15	// The Python wrapper owns only the native indexer handle. The
16	// underlying RocksDB instance remains manager-owned and may continue to
17	// live process-wide for the same .dftindex path.
UNCOV 18	dft_indexer_destroy(self->handle);	×
NEW 19	self->handle = NULL;	×
20	}
21	Py_XDECREF(self->gz_path);	130✔
22	Py_XDECREF(self->index_path);	130✔
23	Py_XDECREF(self->runtime_obj);	130✔
24	Py_TYPE(self)->tp_free((PyObject *)self);	130✔
25	}	130✔
26
27	static void Indexer_release_handle(IndexerObject *self) {	128✔
28	if (self->handle) {	128✔
29	// Releasing the handle drops this wrapper's native indexer state only.
30	// Shared RocksDB lifetime is managed separately by RocksDBManager.
31	dft_indexer_destroy(self->handle);	128✔
32	self->handle = NULL;	128✔
33	}	64✔
34	}	128✔
35
36	static PyObject Indexer_new(PyTypeObject type, PyObject *args,	130✔
37	PyObject *kwds) {
38	IndexerObject *self;
39	self = (IndexerObject *)type->tp_alloc(type, 0);	130✔
40	if (self != NULL) {	130✔
41	self->handle = NULL;	130✔
42	self->gz_path = NULL;	130✔
43	self->index_path = NULL;	130✔
44	self->checkpoint_size = 0;	130✔
45	self->build_bloom = 0;	130✔
46	self->build_manifest = 0;	130✔
47	self->index_threshold =	130✔
48	dftracer::utils::constants::indexer::DEFAULT_INDEX_SIZE_THRESHOLD;
49	self->runtime_obj = NULL;	130✔
50	}	65✔
51	return (PyObject *)self;	130✔
52	}
53
54	static int Indexer_init(IndexerObject self, PyObject args, PyObject *kwds) {	130✔
55	static const char *kwlist[] = {
56	"gz_path", "index_path", "checkpoint_size",
57	"force_rebuild", "build_bloom", "build_manifest",
58	"index_threshold", "runtime", NULL};
59	const char *gz_path;
60	const char *index_path = NULL;	130✔
61	std::uint64_t checkpoint_size =	130✔
62	dftracer::utils::constants::indexer::DEFAULT_CHECKPOINT_SIZE;
63	int force_rebuild = 0;	130✔
64	int build_bloom = 0;	130✔
65	int build_manifest = 0;	130✔
66	std::uint64_t index_threshold =	130✔
67	dftracer::utils::constants::indexer::DEFAULT_INDEX_SIZE_THRESHOLD;
68	PyObject *runtime_arg = NULL;	130✔
69
70	if (!PyArg_ParseTupleAndKeywords(	130!
71	args, kwds, "s\|snpppnO", (char **)kwlist, &gz_path, &index_path,	65✔
72	&checkpoint_size, &force_rebuild, &build_bloom, &build_manifest,
73	&index_threshold, &runtime_arg)) {
74	return -1;	×
75	}
76
77	if (runtime_arg && runtime_arg != Py_None) {	130!
78	if (PyObject_TypeCheck(runtime_arg, &RuntimeType)) {	×
79	Py_INCREF(runtime_arg);	×
80	self->runtime_obj = runtime_arg;	×
81	} else {
82	PyObject *native = PyObject_GetAttrString(runtime_arg, "_native");	×
83	if (native && PyObject_TypeCheck(native, &RuntimeType)) {	×
84	self->runtime_obj = native;	×
85	} else {
86	Py_XDECREF(native);	×
87	PyErr_SetString(PyExc_TypeError,	×
88	"runtime must be a Runtime instance or None");
89	return -1;	×
90	}
91	}
92	}
93
94	self->gz_path = PyUnicode_FromString(gz_path);	130!
95	if (!self->gz_path) {	130✔
96	return -1;	×
97	}
98
99	if (index_path) {	130✔
100	self->index_path = PyUnicode_FromString(index_path);	112!
101	} else {	56✔
102	const std::string index_path = dftracer::utils::utilities::composites::	9!
103	dft::internal::determine_index_path(gz_path, "");	27!
104	self->index_path = PyUnicode_FromString(index_path.c_str());	18!
105	}	18✔
106
107	if (!self->index_path) {	130✔
108	Py_DECREF(self->gz_path);	×
109	return -1;	×
110	}
111
112	self->checkpoint_size = checkpoint_size;	130✔
113	self->build_bloom = build_bloom;	130✔
114	self->build_manifest = build_manifest;	130✔
115	self->index_threshold = index_threshold;	130✔
116
117	const char *index_path_str = PyUnicode_AsUTF8(self->index_path);	130!
118	if (!index_path_str) {	130✔
UNCOV 119	return -1;	×
120	}
121
122	self->handle = dft_indexer_create(gz_path, index_path_str, checkpoint_size,	195!
123	force_rebuild);	65✔
124	if (!self->handle) {	130✔
125	PyErr_SetString(PyExc_RuntimeError, "Failed to create indexer");	2!
126	return -1;	2✔
127	}
128
129	return 0;	128✔
130	}	65✔
131
132	static dftracer::utils::Runtime get_indexer_runtime(IndexerObject self) {	116✔
133	if (self->runtime_obj) {	116!
134	return ((RuntimeObject *)self->runtime_obj)->runtime.get();	×
135	}
136	return get_default_runtime();	116✔
137	}	58✔
138
139	static PyObject Indexer_build(IndexerObject self,	116✔
140	PyObject *Py_UNUSED(ignored)) {
141	if (!self->handle) {	116✔
142	PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");	×
143	return NULL;	×
144	}
145
146	using namespace dftracer::utils;
147	using namespace dftracer::utils::utilities::indexer;
148
149	const char *gz = PyUnicode_AsUTF8(self->gz_path);	116!
150	const char *idx = PyUnicode_AsUTF8(self->index_path);	116!
151	if (!gz \|\| !idx) {	116!
152	return NULL;	×
153	}
154
155	auto config = IndexBuildConfig::for_file(gz)	232!
156	.with_checkpoint_size(	174!
157	static_cast<std::size_t>(self->checkpoint_size))	116!
158	.with_bloom(self->build_bloom != 0)	116!
159	.with_manifest(self->build_manifest != 0)	116!
160	.with_index_threshold(	174!
161	static_cast<std::size_t>(self->index_threshold));	116!
162
163	std::string idx_str(idx);	116!
164	auto pos = idx_str.find_last_of('/');	116✔
165	if (pos != std::string::npos) {	116!
166	config.with_index_dir(idx_str.substr(0, pos));	116!
167	}	58✔
168
169	Runtime *rt = get_indexer_runtime(self);	116!
170	IndexBuildResult build_result;	116✔
171
172	try {
173	auto build_coro =	58✔
174	[](IndexBuildConfig cfg) -> coro::CoroTask<IndexBuildResult> {	464!
175	IndexBuilderUtility builder;	174!
176	co_return co_await builder.process(cfg);	290!
177	};	290!
178
179	Py_BEGIN_ALLOW_THREADS auto handle =	116!
180	rt->submit(build_coro(config), "indexer-build");	174!
181	build_result = handle.get();	116!
182	Py_END_ALLOW_THREADS	116!
183	} catch (const std::exception &e) {	58!
184	PyErr_SetString(PyExc_RuntimeError, e.what());	×
185	return NULL;	×
186	}	×
187
188	if (!build_result.success) {	116✔
189	PyErr_SetString(PyExc_RuntimeError, build_result.error_message.c_str());	×
190	return NULL;	×
191	}
192
193	Py_RETURN_NONE;	116✔
194	}	116✔
195
196	static PyObject Indexer_need_rebuild(IndexerObject self,	42✔
197	PyObject *Py_UNUSED(ignored)) {
198	if (!self->handle) {	42✔
199	PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");	×
200	return NULL;	×
201	}
202
203	int result = dft_indexer_need_rebuild(self->handle);	42✔
204	return PyBool_FromLong(result);	42✔
205	}	21✔
206
207	static PyObject Indexer_exists(IndexerObject self,	×
208	PyObject *Py_UNUSED(ignored)) {
209	if (!self->handle) {	×
210	PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");	×
211	return NULL;	×
212	}
213
214	int result = dft_indexer_exists(self->handle);	×
215	return PyBool_FromLong(result);	×
216	}
217
218	static PyObject Indexer_get_max_bytes(IndexerObject self,	6✔
219	PyObject *Py_UNUSED(ignored)) {
220	if (!self->handle) {	6✔
221	PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");	×
222	return NULL;	×
223	}
224
225	uint64_t result = dft_indexer_get_max_bytes(self->handle);	6✔
226	return PyLong_FromUnsignedLongLong(result);	6✔
227	}	3✔
228
229	static PyObject Indexer_get_num_lines(IndexerObject self,	8✔
230	PyObject *Py_UNUSED(ignored)) {
231	if (!self->handle) {	8✔
232	PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");	×
233	return NULL;	×
234	}
235
236	uint64_t result = dft_indexer_get_num_lines(self->handle);	8✔
237	return PyLong_FromUnsignedLongLong(result);	8✔
238	}	4✔
239
240	static PyObject Indexer_find_checkpoint(IndexerObject self, PyObject *args) {	6✔
241	if (!self->handle) {	6✔
242	PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");	×
243	return NULL;	×
244	}
245
246	std::size_t target_offset;
247	if (!PyArg_ParseTuple(args, "n", &target_offset)) {	6!
248	return NULL;	×
249	}
250
251	dft_indexer_checkpoint_t checkpoint;
252	int found =	3✔
253	dft_indexer_find_checkpoint(self->handle, target_offset, &checkpoint);	6!
254
255	if (!found) {	6✔
256	Py_RETURN_NONE;	2✔
257	}
258
259	// Create IndexerCheckpoint object
260	IndexerCheckpointObject *cp_obj =	2✔
261	(IndexerCheckpointObject *)IndexerCheckpoint_new(&IndexerCheckpointType,	4!
262	NULL, NULL);
263	if (!cp_obj) {	4✔
264	return NULL;	×
265	}
266
267	cp_obj->checkpoint = checkpoint;	4✔
268	return (PyObject *)cp_obj;	4✔
269	}	3✔
270
271	static PyObject Indexer_get_checkpoints(IndexerObject self,	4✔
272	PyObject *Py_UNUSED(ignored)) {
273	if (!self->handle) {	4✔
274	PyErr_SetString(PyExc_RuntimeError, "Indexer not initialized");	×
275	return NULL;	×
276	}
277
278	dft_indexer_checkpoint_t *checkpoints = NULL;	4✔
279	std::size_t count = 0;	4✔
280
281	int result =	2✔
282	dft_indexer_get_checkpoints(self->handle, &checkpoints, &count);	4!
283	if (result != 0 \|\| !checkpoints) {	4!
284	dft_indexer_free_checkpoints(checkpoints, count);	×
285	PyObject *list = PyList_New(0);	×
286	return list;	×
287	}
288
289	PyObject *list = PyList_New(count);	4!
290	if (!list) {	4✔
291	dft_indexer_free_checkpoints(checkpoints, count);	×
292	return NULL;	×
293	}
294
295	for (std::size_t i = 0; i < count; i++) {	170✔
296	IndexerCheckpointObject *cp_obj =	83✔
297	(IndexerCheckpointObject *)IndexerCheckpoint_new(	166!
298	&IndexerCheckpointType, NULL, NULL);
299	if (!cp_obj) {	166!
300	Py_DECREF(list);
301	dft_indexer_free_checkpoints(checkpoints, count);	×
302	return NULL;	×
303	}
304	cp_obj->checkpoint = checkpoints[i];	166✔
305	PyList_SetItem(list, i, (PyObject *)cp_obj);	166!
306	}	83✔
307
308	dft_indexer_free_checkpoints(checkpoints, count);	4!
309	return list;	4✔
310	}	2✔
311
312	static PyObject Indexer_has_bloom(IndexerObject self, void *closure) {	32✔
313	const char *idx = PyUnicode_AsUTF8(self->index_path);	32✔
314	const char *gz = PyUnicode_AsUTF8(self->gz_path);	32✔
315	if (!idx \|\| !gz) {	32!
316	Py_RETURN_FALSE;	×
317	}
318	try {
319	using namespace dftracer::utils::utilities::indexer;
320	using namespace dftracer::utils::utilities::indexer::internal;
321	IndexDatabase db(	16!
322	idx, dftracer::utils::rocksdb::RocksDatabase::OpenMode::ReadOnly);	48!
323	std::string logical = get_logical_path(gz);	32!
324	int fid = db.get_file_info_id(logical);	32!
325	if (fid >= 0 && db.has_bloom_data(fid)) {	32!
326	Py_RETURN_TRUE;	10✔
327	}
328	} catch (...) {	37✔
329	}	×
330	Py_RETURN_FALSE;	22✔
331	}	16✔
332
333	static PyObject Indexer_has_manifest(IndexerObject self, void *closure) {	20✔
334	const char *idx = PyUnicode_AsUTF8(self->index_path);	20✔
335	const char *gz = PyUnicode_AsUTF8(self->gz_path);	20✔
336	if (!idx \|\| !gz) {	20!
337	Py_RETURN_FALSE;	×
338	}
339	try {
340	using namespace dftracer::utils::utilities::indexer;
341	using namespace dftracer::utils::utilities::indexer::internal;
342	IndexDatabase db(	10!
343	idx, dftracer::utils::rocksdb::RocksDatabase::OpenMode::ReadOnly);	30!
344	std::string logical = get_logical_path(gz);	20!
345	int fid = db.get_file_info_id(logical);	20!
346	if (fid >= 0 && db.has_manifest_data(fid)) {	20!
347	Py_RETURN_TRUE;	10✔
348	}
349	} catch (...) {	25✔
350	}	×
351	Py_RETURN_FALSE;	10✔
352	}	10✔
353
354	static PyObject Indexer_gz_path(IndexerObject self, void *closure) {	4✔
355	Py_INCREF(self->gz_path);	4!
356	return self->gz_path;	4✔
357	}
358
359	static PyObject Indexer_index_path(IndexerObject self, void *closure) {	4✔
360	Py_INCREF(self->index_path);	4!
361	return self->index_path;	4✔
362	}
363
364	static PyObject Indexer_checkpoint_size(IndexerObject self, void *closure) {	6✔
365	return PyLong_FromUnsignedLongLong(self->checkpoint_size);	6✔
366	}
367
368	static PyObject Indexer_enter(IndexerObject self,	126!
369	PyObject *Py_UNUSED(ignored)) {
370	Py_INCREF(self);	63✔
371	return (PyObject *)self;	126✔
372	}
373
374	static PyObject Indexer_close(IndexerObject self,	2✔
375	PyObject *Py_UNUSED(ignored)) {
376	Indexer_release_handle(self);	2✔
377	Py_RETURN_NONE;	2✔
378	}
379
380	static PyObject Indexer_exit(IndexerObject self, PyObject *args) {	126✔
381	Indexer_release_handle(self);	126✔
382	Py_RETURN_NONE;	126✔
383	}
384
385	static PyMethodDef Indexer_methods[] = {
386	{"build", (PyCFunction)Indexer_build, METH_NOARGS,
387	"build()\n"
388	"--\n"
389	"\n"
390	"Build or rebuild the index.\n"},
391	{"need_rebuild", (PyCFunction)Indexer_need_rebuild, METH_NOARGS,
392	"Check if a rebuild is needed."},
393	{"exists", (PyCFunction)Indexer_exists, METH_NOARGS,
394	"Check if the .dftindex store exists."},
395	{"get_max_bytes", (PyCFunction)Indexer_get_max_bytes, METH_NOARGS,
396	"Get the maximum uncompressed bytes in the indexed file."},
397	{"get_num_lines", (PyCFunction)Indexer_get_num_lines, METH_NOARGS,
398	"Get the total number of lines in the indexed file."},
399	{"find_checkpoint", (PyCFunction)Indexer_find_checkpoint, METH_VARARGS,
400	"Find the best checkpoint for a given uncompressed offset.\n"
401	"\n"
402	"Args:\n"
403	" offset (int): Uncompressed byte offset.\n"},
404	{"get_checkpoints", (PyCFunction)Indexer_get_checkpoints, METH_NOARGS,
405	"Get all checkpoints for this file as a list."},
406	{"close", (PyCFunction)Indexer_close, METH_NOARGS,
407	"Release this Python wrapper's native indexer handle.\n"
408	"\n"
409	"The shared RocksDB instance for the same .dftindex path remains managed\n"
410	"by the native RocksDBManager cache."},
411	{"__enter__", (PyCFunction)Indexer_enter, METH_NOARGS,
412	"Enter the runtime context for the with statement."},
413	{"__exit__", (PyCFunction)Indexer_exit, METH_VARARGS,
414	"Release this Python wrapper on context exit.\n"
415	"\n"
416	"This does not force-close the shared RocksDB instance for the same\n"
417	".dftindex path."},
418	{NULL} /* Sentinel */
419	};
420
421	static PyGetSetDef Indexer_getsetters[] = {
422	{"gz_path", (getter)Indexer_gz_path, NULL, "Path to the gzip file", NULL},
423	{"index_path", (getter)Indexer_index_path, NULL,
424	"Path to the .dftindex store", NULL},
425	{"checkpoint_size", (getter)Indexer_checkpoint_size, NULL,
426	"Checkpoint size in bytes", NULL},
427	{"has_bloom", (getter)Indexer_has_bloom, NULL,
428	"Whether bloom data exists in index", NULL},
429	{"has_manifest", (getter)Indexer_has_manifest, NULL,
430	"Whether manifest data exists in index", NULL},
431	{NULL} /* Sentinel */
432	};
433
434	PyTypeObject IndexerType = {
435	PyVarObject_HEAD_INIT(NULL, 0) "indexer.Indexer", /* tp_name */
436	sizeof(IndexerObject), /* tp_basicsize */
437	0, /* tp_itemsize */
438	(destructor)Indexer_dealloc, /* tp_dealloc */
439	0, /* tp_vectorcall_offset */
440	0, /* tp_getattr */
441	0, /* tp_setattr */
442	0, /* tp_as_async */
443	0, /* tp_repr */
444	0, /* tp_as_number */
445	0, /* tp_as_sequence */
446	0, /* tp_as_mapping */
447	0, /* tp_hash */
448	0, /* tp_call */
449	0, /* tp_str */
450	0, /* tp_getattro */
451	0, /* tp_setattro */
452	0, /* tp_as_buffer */
453	Py_TPFLAGS_DEFAULT \| Py_TPFLAGS_BASETYPE, /* tp_flags */
454	"Indexer(gz_path: str, index_path: str \| None = None,\n"
455	" checkpoint_size: int = 1048576,\n"
456	" force_rebuild: bool = False, build_bloom: bool = False,\n"
457	" build_manifest: bool = False,\n"
458	" index_threshold: int = 1048576,\n"
459	" runtime: Runtime \| None = None)\n"
460	"--\n"
461	"\n"
462	"Indexer for creating and managing gzip trace index stores.\n"
463	"\n"
464	"Args:\n"
465	" gz_path (str): Path to the gzip trace file.\n"
466	" index_path (str or None): Path to the .dftindex store. If None,\n"
467	" uses the root-local \".dftindex\" next to gz_path.\n"
468	" checkpoint_size (int): Checkpoint size in bytes for index\n"
469	" building (default 1 MB).\n"
470	" force_rebuild (bool): If True, rebuild the index even if it\n"
471	" exists.\n"
472	" build_bloom (bool): If True, build bloom filter data in the\n"
473	" store.\n"
474	" build_manifest (bool): If True, build manifest data in the\n"
475	" store.\n"
476	" index_threshold (int): Skip indexing for files smaller than\n"
477	" this (default 1 MB).\n"
478	" runtime (Runtime or None): Runtime instance for thread pool\n"
479	" control. If None, uses the default global Runtime.\n", /* tp_doc */
480	0, /* tp_traverse */
481	0, /* tp_clear */
482	0, /* tp_richcompare */
483	0, /* tp_weaklistoffset */
484	0, /* tp_iter */
485	0, /* tp_iternext */
486	Indexer_methods, /* tp_methods */
487	0, /* tp_members */
488	Indexer_getsetters, /* tp_getset */
489	0, /* tp_base */
490	0, /* tp_dict */
491	0, /* tp_descr_get */
492	0, /* tp_descr_set */
493	0, /* tp_dictoffset */
494	(initproc)Indexer_init, /* tp_init */
495	0, /* tp_alloc */
496	Indexer_new, /* tp_new */
497	};
498
499	int init_indexer(PyObject *m) {	2✔
500	if (PyType_Ready(&IndexerType) < 0) return -1;	2✔
501
502	Py_INCREF(&IndexerType);	1✔
503	if (PyModule_AddObject(m, "Indexer", (PyObject *)&IndexerType) < 0) {	2✔
504	Py_DECREF(&IndexerType);
505	Py_DECREF(m);
506	return -1;	×
507	}
508
509	return 0;	2✔
510	}	1✔

llnl / dftracer-utils / 24057299873

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous