Merge pull request #23 from lsst/tickets/DM-32131

DM-32131: Merge Cassandra branch of APDB

config/apdb-cassandra.py

@@ -0,0 +1,86 @@
+import lsst.dax.apdb.apdbCassandra
+assert type(config)==lsst.dax.apdb.apdbCassandra.ApdbCassandraConfig, 'config is of type %s.%s instead of lsst.dax.apdb.apdbCassandra.ApdbCassandraConfig' % (type(config).__module__, type(config).__name__)
+# Number of months of history to read from DiaSource
+config.read_sources_months=12
+
+# Number of months of history to read from DiaForcedSource
+config.read_forced_sources_months=12
+
+# Location of (YAML) configuration file with standard schema
+config.schema_file='${DAX_APDB_DIR}/data/apdb-schema.yaml'
+
+# Location of (YAML) configuration file with extra schema
+config.extra_schema_file='${DAX_APDB_DIR}/data/apdb-schema-extra.yaml'
+
+# The list of contact points to try connecting for cluster discovery.
+config.contact_points=['127.0.0.1']
+
+# List of internal IP addresses for contact_points.
+config.private_ips=[]
+
+# Default keyspace for operations.
+config.keyspace='apdb'
+
+# Name for consistency level of read operations, default: QUORUM, can be ONE.
+config.read_consistency='QUORUM'
+
+# Name for consistency level of write operations, default: QUORUM, can be ONE.
+config.write_consistency='QUORUM'
+
+# Timeout in seconds for read operations.
+config.read_timeout=120.0
+
+# Timeout in seconds for write operations.
+config.write_timeout=10.0
+
+# Concurrency level for read operations.
+config.read_concurrency=500
+
+# Cassandra protocol version to use, default is V4
+config.protocol_version=4
+
+# List of columns to read from DiaObject, by default read all columns
+config.dia_object_columns=[]
+
+# Prefix to add to table names
+config.prefix=''
+
+# Pixelization used for partitioning index.
+config.part_pixelization='mq3c'
+
+# Pixelization level used for partitioning index.
+config.part_pix_level=10
+
+# Names ra/dec columns in DiaObject table
+config.ra_dec_columns=['ra', 'decl']
+
+# If True then print/log timing information
+config.timer=False
+
+# Use per-partition tables for sources instead of partitioning by time
+config.time_partition_tables=True
+
+# Time partitoning granularity in days, this value must not be changed after database is initialized
+config.time_partition_days=30
+
+# Starting time for per-partion tables, in yyyy-mm-ddThh:mm:ss format, in TAI. This is used only when time_partition_tables is True.
+config.time_partition_start='2018-12-01T00:00:00'
+
+# Ending time for per-partion tables, in yyyy-mm-ddThh:mm:ss format, in TAI This is used only when time_partition_tables is True.
+config.time_partition_end='2030-01-01T00:00:00'
+
+# If True then build separate query for each time partition, otherwise build one single query. This is only used when time_partition_tables is False in schema config.
+config.query_per_time_part=False
+
+# If True then build one query per spacial partition, otherwise build single query.
+config.query_per_spatial_part=False
+
+# If True then combine result rows before converting to pandas.
+config.pandas_delay_conv=True
+
+# Packing method for table records.
+config.packing='none'
+
+# If True use Cassandra prepared statements.
+config.prepared_statements=True
+

data/apdb-schema-extra.yaml

@@ -119,33 +119,8 @@ columns:
   description: Covariance of pmDecl and parallax.
   ucd: stat.covariance
   unit: mas^2/yr
-- name: pixelId
-  type: BIGINT
-  nullable: false
-  description: HTM index.
 indices:
 - name: PK_DiaObjectLast
   columns:
-  - pixelId
-  - diaObjectId
-  type: PRIMARY
-- name: IDX_DiaObjLast_diaObjId
-  columns:
-  - diaObjectId
-  type: INDEX
-
----
-# Special PK index for DiaObject table with spacial column being first
-# (should provide better locality)
-table: DiaObjectIndexHtmFirst
-indices:
-- name: PK_DiaObject
-  columns:
-  - pixelId
   - diaObjectId
-  - validityStart
   type: PRIMARY
-- name: IDX_DiaObject_diaObjId
-  columns:
-  - diaObjectId
-  type: INDEX

data/apdb-schema.yaml

@@ -501,10 +501,6 @@ columns:
   default: '0'
   description: Flags, bitwise OR tbd.
   ucd: meta.code
-- name: pixelId
-  type: BIGINT
-  nullable: false
-  description: HTM index.
 indices:
 - name: PK_DiaObject
   columns:
@@ -515,10 +511,6 @@ indices:
   columns:
   - validityStart
   type: INDEX
-- name: IDX_DiaObject_htmId20
-  columns:
-  - pixelId
-  type: INDEX
 ---
 table: SSObject
 description: The SSObject table contains description of the Solar System (moving)
@@ -1548,10 +1540,6 @@ columns:
   default: '0'
   description: Flags, bitwise OR tbd.
   ucd: meta.code
-- name: pixelId
-  type: BIGINT
-  nullable: false
-  description: HTM index.
 indices:
 - name: PK_DiaSource
   columns:
@@ -1569,10 +1557,6 @@ indices:
   columns:
   - ssObjectId
   type: INDEX
-- name: IDX_DiaSource_htmId20
-  columns:
-  - pixelId
-  type: INDEX
 ---
 table: DiaForcedSource
 description: Forced-photometry source measurement on an individual difference Exposure

mypy.ini

@@ -9,8 +9,11 @@ disallow_incomplete_defs = True
 [mypy-sqlalchemy.*]
 ignore_missing_imports = True
 
-; [mypy-pandas.*]
-; ignore_missing_imports = True
+[mypy-cassandra.*]
+ignore_missing_imports = True
+
+[mypy-cbor.*]
+ignore_missing_imports = True
 
 [mypy-lsst.*]
 ignore_missing_imports = True

python/lsst/dax/apdb/__init__.py

@@ -1,5 +1,9 @@
 
 from .apdb import *
+from .apdbSchema import *
+from .apdbSqlSchema import *
 from .apdbSql import *
+from .apdbCassandraSchema import *
+from .apdbCassandra import *
 from .factory import *
 from .version import *

python/lsst/dax/apdb/apdb.py

@@ -31,6 +31,7 @@
 import lsst.daf.base as dafBase
 from lsst.pex.config import Config, ConfigurableField, Field
 from lsst.sphgeom import Region
+from .apdbSchema import ApdbTables, TableDef
 
 
 def _data_file_name(basename: str) -> str:
@@ -42,18 +43,28 @@ def _data_file_name(basename: str) -> str:
 class ApdbConfig(Config):
     """Part of Apdb configuration common to all implementations.
     """
-    read_sources_months = Field(dtype=int,
-                                doc="Number of months of history to read from DiaSource",
-                                default=12)
-    read_forced_sources_months = Field(dtype=int,
-                                       doc="Number of months of history to read from DiaForcedSource",
-                                       default=12)
-    schema_file = Field(dtype=str,
-                        doc="Location of (YAML) configuration file with standard schema",
-                        default=_data_file_name("apdb-schema.yaml"))
-    extra_schema_file = Field(dtype=str,
-                              doc="Location of (YAML) configuration file with extra schema",
-                              default=_data_file_name("apdb-schema-extra.yaml"))
+    read_sources_months = Field(
+        dtype=int,
+        doc="Number of months of history to read from DiaSource",
+        default=12
+    )
+    read_forced_sources_months = Field(
+        dtype=int,
+        doc="Number of months of history to read from DiaForcedSource",
+        default=12
+    )
+    schema_file = Field(
+        dtype=str,
+        doc="Location of (YAML) configuration file with standard schema",
+        default=_data_file_name("apdb-schema.yaml")
+    )
+    extra_schema_file = Field(
+        dtype=str,
+        doc="Location of (YAML) configuration file with extra schema, "
+            "definitions in this file are merged with the definitions in "
+            "'schema_file', extending or replacing parts of the schema.",
+        default=_data_file_name("apdb-schema-extra.yaml")
+    )
 
 
 class Apdb(ABC):
@@ -62,6 +73,23 @@ class Apdb(ABC):
 
     ConfigClass = ApdbConfig
 
+    @abstractmethod
+    def tableDef(self, table: ApdbTables) -> Optional[TableDef]:
+        """Return table schema definition for a given table.
+
+        Parameters
+        ----------
+        table : `ApdbTables`
+            One of the known APDB tables.
+
+        Returns
+        -------
+        tableSchema : `TableDef` or `None`
+            Table schema description, `None` is returned if table is not
+            defined by this implementation.
+        """
+        raise NotImplementedError()
+
     @abstractmethod
     def makeSchema(self, drop: bool = False) -> None:
         """Create or re-create whole database schema.