Inheritance diagram for lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica:

Public Member Functions
	__init__ (self, ApdbCassandra apdb, ApdbCassandraSchema schema, Any session)

VersionTuple	apdbReplicaImplementationVersion (cls)

list[ReplicaChunk]\|None	getReplicaChunks (self)

None	deleteReplicaChunks (self, Iterable[int] chunks)

ApdbTableData	getDiaObjectsChunks (self, Iterable[int] chunks)

ApdbTableData	getDiaSourcesChunks (self, Iterable[int] chunks)

ApdbTableData	getDiaForcedSourcesChunks (self, Iterable[int] chunks)

Protected Member Functions
Timer	_timer (self, str name, *Mapping[str, str\|int]\|None tags=None)

ApdbTableData	_get_chunks (self, ExtraTables table, Iterable[int] chunks)

Protected Attributes
	_apdb

	_schema

	_session

	_config

	_preparer

	_timer_args

Detailed Description

Implementation of `ApdbReplica` for Cassandra backend.

Parameters
----------
apdb : `ApdbCassandra`
    Instance of ApbdCassandra for database.
schema : `ApdbCassandraSchema`
    Instance of ApdbCassandraSchema for database.
session
    Instance of cassandra session type.

Definition at line 54 of file apdbCassandraReplica.py.

Constructor & Destructor Documentation

◆ init()

lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica.__init__	(		self,
		ApdbCassandra	apdb,
		ApdbCassandraSchema	schema,
		Any	session )

Definition at line 67 of file apdbCassandraReplica.py.

    def __init__(self, apdb: ApdbCassandra, schema: ApdbCassandraSchema, session: Any):
        # Note that ApdbCassandra instance must stay alive while this object
        # exists, so we keep reference to it.
        self._apdb = apdb
        self._schema = schema
        self._session = session
        self._config = apdb.config
 
        # Cache for prepared statements
        self._preparer = PreparedStatementCache(self._session)
 
        self._timer_args: list[MonAgent | logging.Logger] = [_MON]
        if self._config.timer:
            self._timer_args.append(_LOG)
 

Member Function Documentation

◆ _get_chunks()

ApdbTableData lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica._get_chunks	(		self,
		ExtraTables	table,
		Iterable[int]	chunks )

protected

Return records from a particular table given set of insert IDs.

Definition at line 179 of file apdbCassandraReplica.py.

    def _get_chunks(self, table: ExtraTables, chunks: Iterable[int]) -> ApdbTableData:
        """Return records from a particular table given set of insert IDs."""
        if not self._schema.has_replica_chunks:
            raise ValueError("APDB is not configured for replication")
 
        # We do not expect too may chunks in this query.
        chunks = list(chunks)
        params = ",".join("?" * len(chunks))
 
        table_name = self._schema.tableName(table)
        # I know that chunk table schema has only regular APDB columns plus
        # apdb_replica_chunk column, and this is exactly what we need to return
        # from this method, so selecting a star is fine here.
        query = (
            f'SELECT * FROM "{self._config.keyspace}"."{table_name}" WHERE apdb_replica_chunk IN ({params})'
        )
        statement = self._preparer.prepare(query)
 
        with self._timer("table_chunk_select_time", tags={"table": table_name}):
            result = self._session.execute(statement, chunks, execution_profile="read_raw")
            table_data = cast(ApdbCassandraTableData, result._current_rows)
        return table_data

◆ _timer()

Timer lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica._timer	(		self,
		str	name,
		*Mapping[str, str \| int] \| None	tags = None )

protected

Create `Timer` instance given its name.

Definition at line 82 of file apdbCassandraReplica.py.

    def _timer(self, name: str, *, tags: Mapping[str, str | int] | None = None) -> Timer:
        """Create `Timer` instance given its name."""
        return Timer(name, *self._timer_args, tags=tags)
 

◆ apdbReplicaImplementationVersion()

VersionTuple lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica.apdbReplicaImplementationVersion ( cls )

Return version number for current ApdbReplica implementation.

Returns
-------
version : `VersionTuple`
    Version of the code defined in implementation class.

Reimplemented from lsst.dax.apdb.apdbReplica.ApdbReplica.

Definition at line 87 of file apdbCassandraReplica.py.

    def apdbReplicaImplementationVersion(cls) -> VersionTuple:
        # Docstring inherited from base class.
        return VERSION
 

◆ deleteReplicaChunks()

None lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica.deleteReplicaChunks	(		self,
		Iterable[int]	chunks )

Remove replication chunks from the database.

Parameters
----------
chunks : `~collections.abc.Iterable` [`int`]
    Chunk identifiers to remove.

Notes
-----
This method causes Apdb to forget about specified chunks. If there
are any auxiliary data associated with the identifiers, it is also
removed from database (but data in regular tables is not removed).
This method should be called after successful transfer of data from
APDB to PPDB to free space used by replicas.

Reimplemented from lsst.dax.apdb.apdbReplica.ApdbReplica.

Definition at line 124 of file apdbCassandraReplica.py.

    def deleteReplicaChunks(self, chunks: Iterable[int]) -> None:
        # docstring is inherited from a base class
        if not self._schema.has_replica_chunks:
            raise ValueError("APDB is not configured for replication")
 
        # There is 64k limit on number of markers in Cassandra CQL
        for chunk_ids in chunk_iterable(chunks, 20_000):
            params = ",".join("?" * len(chunk_ids))
 
            # everything goes into a single partition
            partition = 0
 
            table_name = self._schema.tableName(ExtraTables.ApdbReplicaChunks)
            query = (
                f'DELETE FROM "{self._config.keyspace}"."{table_name}" '
                f"WHERE partition = ? AND apdb_replica_chunk IN ({params})"
            )
 
            with self._timer("chunks_delete_time"):
                self._session.execute(
                    self._preparer.prepare(query),
                    [partition] + list(chunk_ids),
                    timeout=self._config.remove_timeout,
                )
 
            # Also remove those chunk_ids from Dia*Chunks tables.
            for table in (
                ExtraTables.DiaObjectChunks,
                ExtraTables.DiaSourceChunks,
                ExtraTables.DiaForcedSourceChunks,
            ):
                table_name = self._schema.tableName(table)
                query = (
                    f'DELETE FROM "{self._config.keyspace}"."{table_name}"'
                    f" WHERE apdb_replica_chunk IN ({params})"
                )
                with self._timer("table_chunk_detele_time", tags={"table": table_name}):
                    self._session.execute(
                        self._preparer.prepare(query),
                        chunk_ids,
                        timeout=self._config.remove_timeout,
                    )
 

◆ getDiaForcedSourcesChunks()

ApdbTableData lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica.getDiaForcedSourcesChunks	(		self,
		Iterable[int]	chunks )

Return catalog of DiaForcedSource records from given replica chunks.

Parameters
----------
chunks : `~collections.abc.Iterable` [`int`]
    Chunk identifiers to return.

Returns
-------
data : `ApdbTableData`
    Catalog containing DiaForcedSource records. In addition to all
    regular columns it will contain ``apdb_replica_chunk`` column.

Notes
-----
This part of API may not be very stable and can change before the
implementation finalizes.

Reimplemented from lsst.dax.apdb.apdbReplica.ApdbReplica.

Definition at line 175 of file apdbCassandraReplica.py.

    def getDiaForcedSourcesChunks(self, chunks: Iterable[int]) -> ApdbTableData:
        # docstring is inherited from a base class
        return self._get_chunks(ExtraTables.DiaForcedSourceChunks, chunks)
 

◆ getDiaObjectsChunks()

ApdbTableData lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica.getDiaObjectsChunks	(		self,
		Iterable[int]	chunks )

Return catalog of DiaObject records from given replica chunks.

Parameters
----------
chunks : `~collections.abc.Iterable` [`int`]
    Chunk identifiers to return.

Returns
-------
data : `ApdbTableData`
    Catalog containing DiaObject records. In addition to all regular
    columns it will contain ``apdb_replica_chunk`` column.

Notes
-----
This part of API may not be very stable and can change before the
implementation finalizes.

Reimplemented from lsst.dax.apdb.apdbReplica.ApdbReplica.

Definition at line 167 of file apdbCassandraReplica.py.

    def getDiaObjectsChunks(self, chunks: Iterable[int]) -> ApdbTableData:
        # docstring is inherited from a base class
        return self._get_chunks(ExtraTables.DiaObjectChunks, chunks)
 

◆ getDiaSourcesChunks()

ApdbTableData lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica.getDiaSourcesChunks	(		self,
		Iterable[int]	chunks )

Return catalog of DiaSource records from given replica chunks.

Parameters
----------
chunks : `~collections.abc.Iterable` [`int`]
    Chunk identifiers to return.

Returns
-------
data : `ApdbTableData`
    Catalog containing DiaSource records. In addition to all regular
    columns it will contain ``apdb_replica_chunk`` column.

Notes
-----
This part of API may not be very stable and can change before the
implementation finalizes.

Reimplemented from lsst.dax.apdb.apdbReplica.ApdbReplica.

Definition at line 171 of file apdbCassandraReplica.py.

    def getDiaSourcesChunks(self, chunks: Iterable[int]) -> ApdbTableData:
        # docstring is inherited from a base class
        return self._get_chunks(ExtraTables.DiaSourceChunks, chunks)
 

◆ getReplicaChunks()

list[ReplicaChunk] | None lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica.getReplicaChunks ( self )

Return collection of replication chunks known to the database.

Returns
-------
chunks : `list` [`ReplicaChunk`] or `None`
    List of chunks, they may be time-ordered if database supports
    ordering. `None` is returned if database is not configured for
    replication.

Reimplemented from lsst.dax.apdb.apdbReplica.ApdbReplica.

Definition at line 91 of file apdbCassandraReplica.py.

    def getReplicaChunks(self) -> list[ReplicaChunk] | None:
        # docstring is inherited from a base class
        if not self._schema.has_replica_chunks:
            return None
 
        # everything goes into a single partition
        partition = 0
 
        table_name = self._schema.tableName(ExtraTables.ApdbReplicaChunks)
        # We want to avoid timezone mess so return timestamps as milliseconds.
        query = (
            "SELECT toUnixTimestamp(last_update_time), apdb_replica_chunk, unique_id "
            f'FROM "{self._config.keyspace}"."{table_name}" WHERE partition = ?'
        )
 
        with self._timer("chunks_select_time"):
            result = self._session.execute(
                self._preparer.prepare(query),
                (partition,),
                timeout=self._config.read_timeout,
                execution_profile="read_tuples",
            )
        # order by last_update_time
        rows = sorted(result)
        return [
            ReplicaChunk(
                id=row[1],
                last_update_time=astropy.time.Time(row[0] / 1000, format="unix_tai"),
                unique_id=row[2],
            )
            for row in rows
        ]
 

Member Data Documentation

◆ _apdb

lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica._apdb

protected

Definition at line 70 of file apdbCassandraReplica.py.

◆ _config

lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica._config

protected

Definition at line 73 of file apdbCassandraReplica.py.

◆ _preparer

lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica._preparer

protected

Definition at line 76 of file apdbCassandraReplica.py.

◆ _schema

lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica._schema

protected

Definition at line 71 of file apdbCassandraReplica.py.

◆ _session

lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica._session

protected

Definition at line 72 of file apdbCassandraReplica.py.

◆ _timer_args

lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica._timer_args

protected

Definition at line 84 of file apdbCassandraReplica.py.

The documentation for this class was generated from the following file:

/j/snowflake/release/lsstsw/stack/lsst-scipipe-8.0.0/Linux64/dax_apdb/g88963caddf+0cb8e002cc/python/lsst/dax/apdb/cassandra/apdbCassandraReplica.py

Public Member Functions

Protected Member Functions

Protected Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ _get_chunks()

◆ _timer()

◆ apdbReplicaImplementationVersion()

◆ deleteReplicaChunks()

◆ getDiaForcedSourcesChunks()

◆ getDiaObjectsChunks()

◆ getDiaSourcesChunks()

◆ getReplicaChunks()

Member Data Documentation

◆ _apdb

◆ _config

◆ _preparer

◆ _schema

◆ _session

◆ _timer_args

◆ init()