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

Public Member Functions
	__init__ (self, ApdbCassandra apdb)

VersionTuple	schemaVersion (self)

VersionTuple	apdbReplicaImplementationVersion (cls)

bool	hasChunkSubPartitions (cls, VersionTuple version)

bool	hasUpdateRecordChunks (cls, VersionTuple version)

list[ReplicaChunk]\|None	getReplicaChunks (self)

None	deleteReplicaChunks (self, Iterable[int] chunks)

ApdbTableData	getTableDataChunks (self, ApdbTables table, Iterable[int] chunks)

Sequence[ApdbUpdateRecord]	getUpdateRecordChunks (self, Iterable[int] chunks)

ApdbReplica	from_config (cls, ApdbConfig config)

ApdbReplica	from_uri (cls, ResourcePathExpression uri)

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

Protected Attributes
	_apdb = apdb

Detailed Description

Implementation of `ApdbReplica` for Cassandra backend.

Parameters
----------
apdb : `ApdbCassandra`
    Instance of ApbdCassandra for database.

Definition at line 61 of file apdbCassandraReplica.py.

Constructor & Destructor Documentation

◆ init()

lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica.__init__	(		self,
		ApdbCassandra	apdb )

Definition at line 70 of file apdbCassandraReplica.py.

    def __init__(self, apdb: ApdbCassandra):
        # Note that ApdbCassandra instance must stay alive while this object
        # exists, so we keep reference to it.
        self._apdb = apdb
 

Member Function Documentation

◆ _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 75 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, _MON, 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 85 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 136 of file apdbCassandraReplica.py.

    def deleteReplicaChunks(self, chunks: Iterable[int]) -> None:
        # docstring is inherited from a base class
        context = self._apdb._context
        config = context.config
 
        if not context.schema.replication_enabled:
            raise ValueError("APDB is not configured for replication")
 
        # everything goes into a single partition
        partition = 0
 
        # Iterable can be single pass, make everything that we need from it
        # in a single loop.
        repl_table_params = []
        chunk_table_params: list[tuple] = []
        for chunk in chunks:
            repl_table_params.append((partition, chunk))
            if context.has_chunk_sub_partitions:
                for subchunk in range(config.replica_sub_chunk_count):
                    chunk_table_params.append((chunk, subchunk))
            else:
                chunk_table_params.append((chunk,))
        # Anything to do att all?
        if not repl_table_params:
            return
 
        table_name = context.schema.tableName(ExtraTables.ApdbReplicaChunks)
        query = (
            f'DELETE FROM "{config.keyspace}"."{table_name}" WHERE partition = ? AND apdb_replica_chunk = ?'
        )
        statement = context.preparer.prepare(query)
 
        queries = [(statement, param) for param in repl_table_params]
        with self._timer("chunks_delete_time") as timer:
            execute_concurrent(context.session, queries)
            timer.add_values(row_count=len(queries))
 
        # Also remove those chunk_ids from Dia*Chunks tables.
        tables = list(ExtraTables.replica_chunk_tables(context.has_chunk_sub_partitions).values())
        if context.has_update_record_chunks_table:
            tables.append(ExtraTables.ApdbUpdateRecordChunks)
        for table in tables:
            table_name = context.schema.tableName(table)
            query = f'DELETE FROM "{config.keyspace}"."{table_name}" WHERE apdb_replica_chunk = ?'
            if context.has_chunk_sub_partitions:
                query += " AND apdb_replica_subchunk = ?"
            statement = context.preparer.prepare(query)
 
            queries = [(statement, param) for param in chunk_table_params]
            with self._timer("table_chunk_detele_time", tags={"table": table_name}) as timer:
                execute_concurrent(context.session, queries)
                timer.add_values(row_count=len(queries))
 

◆ from_config()

ApdbReplica lsst.dax.apdb.apdbReplica.ApdbReplica.from_config	(		cls,
		ApdbConfig	config )

inherited

Create ApdbReplica instance from configuration object.

Parameters
----------
config : `ApdbConfig`
    Configuration object, type of this object determines type of the
    ApdbReplica implementation.

Returns
-------
replica : `ApdbReplica`
    Instance of `ApdbReplica` class.

Definition at line 127 of file apdbReplica.py.

    def from_config(cls, config: ApdbConfig) -> ApdbReplica:
        """Create ApdbReplica instance from configuration object.
 
        Parameters
        ----------
        config : `ApdbConfig`
            Configuration object, type of this object determines type of the
            ApdbReplica implementation.
 
        Returns
        -------
        replica : `ApdbReplica`
            Instance of `ApdbReplica` class.
        """
        return make_apdb_replica(config)
 

◆ from_uri()

ApdbReplica lsst.dax.apdb.apdbReplica.ApdbReplica.from_uri	(		cls,
		ResourcePathExpression	uri )

inherited

Make ApdbReplica instance from a serialized configuration.

Parameters
----------
uri : `~lsst.resources.ResourcePathExpression`
    URI or local file path pointing to a file with serialized
    configuration, or a string with a "label:" prefix. In the latter
    case, the configuration will be looked up from an APDB index file
    using the label name that follows the prefix. The APDB index file's
    location is determined by the ``DAX_APDB_INDEX_URI`` environment
    variable.

Returns
-------
replica : `ApdbReplica`
    Instance of `ApdbReplica` class, the type of the returned instance
    is determined by configuration.

Definition at line 144 of file apdbReplica.py.

    def from_uri(cls, uri: ResourcePathExpression) -> ApdbReplica:
        """Make ApdbReplica instance from a serialized configuration.
 
        Parameters
        ----------
        uri : `~lsst.resources.ResourcePathExpression`
            URI or local file path pointing to a file with serialized
            configuration, or a string with a "label:" prefix. In the latter
            case, the configuration will be looked up from an APDB index file
            using the label name that follows the prefix. The APDB index file's
            location is determined by the ``DAX_APDB_INDEX_URI`` environment
            variable.
 
        Returns
        -------
        replica : `ApdbReplica`
            Instance of `ApdbReplica` class, the type of the returned instance
            is determined by configuration.
        """
        config = ApdbConfig.from_uri(uri)
        return make_apdb_replica(config)
 

◆ 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 99 of file apdbCassandraReplica.py.

    def getReplicaChunks(self) -> list[ReplicaChunk] | None:
        # docstring is inherited from a base class
        context = self._apdb._context
        config = context.config
 
        if not context.schema.replication_enabled:
            return None
 
        # everything goes into a single partition
        partition = 0
 
        table_name = context.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 "{config.keyspace}"."{table_name}" WHERE partition = %s'
        )
 
        with self._timer("chunks_select_time") as timer:
            result = context.session.execute(
                query,
                (partition,),
                timeout=config.connection_config.read_timeout,
                execution_profile="read_tuples",
            )
            # order by last_update_time
            rows = sorted(result)
            timer.add_values(row_count=len(rows))
        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
        ]
 

◆ getTableDataChunks()

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

Return catalog of new records for a table from given replica chunks.

Parameters
----------
table : `ApdbTables`
    Table for which to return the data. Acceptable tables are
    `ApdbTables.DiaObject`, `ApdbTables.DiaSource`, and
    `ApdbTables.DiaForcedSource`.
chunks : `~collections.abc.Iterable` [`int`]
    Chunk identifiers to return.

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

Notes
-----
This method returns new records that have been added to the table by
`Apdb.store()` method. Updates to the records that happen at later time
are available from `getTableUpdateChunks` method.

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 189 of file apdbCassandraReplica.py.

    def getTableDataChunks(self, table: ApdbTables, chunks: Iterable[int]) -> ApdbTableData:
        # docstring is inherited from a base class
        context = self._apdb._context
        config = context.config
 
        if not context.schema.replication_enabled:
            raise ValueError("APDB is not configured for replication")
        if table not in ExtraTables.replica_chunk_tables(False):
            raise ValueError(f"Table {table} does not support replica chunks.")
 
        # We need to iterate few times.
        chunks = list(chunks)
 
        # If schema was migrated then a chunk can appear in either old or new
        # chunk table (e.g. DiaObjectChunks or DiaObjectChunks2). Chunk table
        # has a column which will be set to true for new table.
        has_chunk_sub_partitions: dict[int, bool] = {}
        if context.has_chunk_sub_partitions:
            table_name = context.schema.tableName(ExtraTables.ApdbReplicaChunks)
            chunks_str = ",".join(str(chunk_id) for chunk_id in chunks)
            query = (
                f'SELECT apdb_replica_chunk, has_subchunks FROM "{config.keyspace}"."{table_name}" '
                f"WHERE partition = %s and apdb_replica_chunk IN ({chunks_str})"
            )
            partition = 0
            result = context.session.execute(
                query,
                (partition,),
                timeout=config.connection_config.read_timeout,
                execution_profile="read_tuples",
            )
            has_chunk_sub_partitions = dict(result)
        else:
            has_chunk_sub_partitions = dict.fromkeys(chunks, False)
 
        # Check what kind of tables we want to query, if chunk list is empty
        # then use tables which should exist in the schema.
        if has_chunk_sub_partitions:
            have_subchunks = any(has_chunk_sub_partitions.values())
            have_non_subchunks = not all(has_chunk_sub_partitions.values())
        else:
            have_subchunks = context.has_chunk_sub_partitions
            have_non_subchunks = not have_subchunks
 
        # NOTE: if an existing database is migrated and has both types of chunk
        # tables (e.g. DiaObjectChunks and DiaObjectChunks2) it is possible
        # that the same chunk can appear in both tables. In reality schema
        # migration should only happen during the downtime, so there will be
        # suffient gap and a different chunk ID will be used for new chunks.
 
        table_data: ApdbCassandraTableData | None = None
        table_data_subchunk: ApdbCassandraTableData | None = None
 
        table_name = context.schema.tableName(ExtraTables.replica_chunk_tables(False)[table])
        with self._timer("table_chunk_select_time", tags={"table": table_name}) as timer:
            if have_subchunks:
                replica_table = ExtraTables.replica_chunk_tables(True)[table]
                table_name = context.schema.tableName(replica_table)
                query = (
                    f'SELECT * FROM "{config.keyspace}"."{table_name}" '
                    "WHERE apdb_replica_chunk = ? AND apdb_replica_subchunk = ?"
                )
                statement = context.preparer.prepare(query)
 
                queries: list[tuple] = []
                for chunk in chunks:
                    if has_chunk_sub_partitions.get(chunk, False):
                        for subchunk in range(config.replica_sub_chunk_count):
                            queries.append((statement, (chunk, subchunk)))
                if not queries and not have_non_subchunks:
                    # Add a dummy query to return correct set of columns.
                    queries.append((statement, (-1, -1)))
 
                if queries:
                    table_data_subchunk = cast(
                        ApdbCassandraTableData,
                        select_concurrent(
                            context.session,
                            queries,
                            "read_raw_multi",
                            config.connection_config.read_concurrency,
                        ),
                    )
 
            if have_non_subchunks:
                replica_table = ExtraTables.replica_chunk_tables(False)[table]
                table_name = context.schema.tableName(replica_table)
                query = f'SELECT * FROM "{config.keyspace}"."{table_name}" WHERE apdb_replica_chunk = ?'
                statement = context.preparer.prepare(query)
 
                queries = []
                for chunk in chunks:
                    if not has_chunk_sub_partitions.get(chunk, True):
                        queries.append((statement, (chunk,)))
                if not queries and not table_data_subchunk:
                    # Add a dummy query to return correct set of columns.
                    queries.append((statement, (-1,)))
 
                if queries:
                    table_data = cast(
                        ApdbCassandraTableData,
                        select_concurrent(
                            context.session,
                            queries,
                            "read_raw_multi",
                            config.connection_config.read_concurrency,
                        ),
                    )
 
            # Merge if both are non-empty.
            if table_data and table_data_subchunk:
                table_data_subchunk.project(drop=["apdb_replica_subchunk"])
                table_data.append(table_data_subchunk)
            elif table_data_subchunk:
                table_data = table_data_subchunk
            elif not table_data:
                raise AssertionError("above logic is incorrect")
 
            timer.add_values(row_count=len(table_data.rows()))
 
            table_schema = self._apdb._schema.tableSchemas[table]
            # Regular tables should never have columns of ExtraDataTypes, this
            # is just to make mypy happy.
            column_types = {
                column.name: column.datatype
                for column in table_schema.columns
                if not isinstance(column.datatype, ExtraDataTypes)
            }
            column_types["apdb_replica_chunk"] = felis.datamodel.DataType.long
            # It may also have subchunk column, we do not always drop it, and
            # clients should not need it, but we need to provide type for it.
            column_types["apdb_replica_subchunk"] = felis.datamodel.DataType.int
            table_data.set_column_types(column_types)
 
        return table_data
 

◆ getUpdateRecordChunks()

Sequence[ApdbUpdateRecord] lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica.getUpdateRecordChunks	(		self,
		Iterable[int]	chunks )

Return the list of record updates from given replica chunks.

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

Returns
-------
records : `~collections.abc.Sequence` [`ApdbUpdateRecord`]
    Collection of update records. Records will be sorted according
    their update time and update order.

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

Definition at line 325 of file apdbCassandraReplica.py.

    def getUpdateRecordChunks(self, chunks: Iterable[int]) -> Sequence[ApdbUpdateRecord]:
        # docstring is inherited from a base class
        context = self._apdb._context
        config = context.config
 
        if not context.schema.replication_enabled:
            raise ValueError("APDB is not configured for replication")
 
        if not context.has_update_record_chunks_table:
            # Table does not exist yet.
            return []
 
        table_name = context.schema.tableName(ExtraTables.ApdbUpdateRecordChunks)
 
        records = []
        if context.has_chunk_sub_partitions:
            subchunks = ",".join(str(val) for val in range(config.replica_sub_chunk_count))
            query = (
                f'SELECT * FROM "{config.keyspace}"."{table_name}" '
                f"WHERE apdb_replica_chunk = %s AND apdb_replica_subchunk IN ({subchunks})"
            )
 
            with self._timer("select_update_record_time", tags={"table": table_name}) as timer:
                for chunk in chunks:
                    result = context.session.execute(query, [chunk])
                    for row in result:
                        records.append(
                            ApdbUpdateRecord.from_json(
                                row.update_time_ns, row.update_order, row.update_payload
                            )
                        )
                timer.add_values(row_count=len(records))
 
        else:
            chunks_str = ",".join(str(val) for val in chunks)
            query = (
                f'SELECT * FROM "{config.keyspace}"."{table_name}" WHERE apdb_replica_chunk IN ({chunks_str})'
            )
 
            with self._timer("select_update_record_time", tags={"table": table_name}) as timer:
                result = context.session.execute(query)
                for row in result:
                    records.append(
                        ApdbUpdateRecord.from_json(row.update_time_ns, row.update_order, row.update_payload)
                    )
                timer.add_values(row_count=len(records))
 
        records.sort()
        return records

◆ hasChunkSubPartitions()

bool lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica.hasChunkSubPartitions	(		cls,
		VersionTuple	version )

Return True if replica chunk tables have sub-partitions.

Definition at line 90 of file apdbCassandraReplica.py.

    def hasChunkSubPartitions(cls, version: VersionTuple) -> bool:
        """Return True if replica chunk tables have sub-partitions."""
        return version >= VersionTuple(1, 1, 0)
 

◆ hasUpdateRecordChunks()

bool lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica.hasUpdateRecordChunks	(		cls,
		VersionTuple	version )

Return True if ApdbUpdateRecordChunks should exists.

Definition at line 95 of file apdbCassandraReplica.py.

    def hasUpdateRecordChunks(cls, version: VersionTuple) -> bool:
        """Return True if ApdbUpdateRecordChunks should exists."""
        return version >= VersionTuple(1, 1, 1)
 

◆ schemaVersion()

VersionTuple lsst.dax.apdb.cassandra.apdbCassandraReplica.ApdbCassandraReplica.schemaVersion ( self )

Return version number of the database schema.

Returns
-------
version : `VersionTuple`
    Version of the database schema.

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

Definition at line 79 of file apdbCassandraReplica.py.

    def schemaVersion(self) -> VersionTuple:
        # Docstring inherited from base class.
        context = self._apdb._context
        return context.db_versions.schema_version
 

Member Data Documentation

◆ _apdb

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

protected

Definition at line 73 of file apdbCassandraReplica.py.

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

/j/snowflake/release/lsstsw/stack/lsst-scipipe-12.1.0/Linux/dax_apdb/gb88ae4c679+7dec8f19df/python/lsst/dax/apdb/cassandra/apdbCassandraReplica.py

Public Member Functions

Protected Member Functions

Protected Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ _timer()

◆ apdbReplicaImplementationVersion()

◆ deleteReplicaChunks()

◆ from_config()

◆ from_uri()

◆ getReplicaChunks()

◆ getTableDataChunks()

◆ getUpdateRecordChunks()

◆ hasChunkSubPartitions()

◆ hasUpdateRecordChunks()

◆ schemaVersion()

Member Data Documentation

◆ _apdb

◆ init()