misc.python.materialize.zippy.view_capabilities 

 1# Copyright Materialize, Inc. and contributors. All rights reserved.
 2#
 3# Use of this software is governed by the Business Source License
 4# included in the LICENSE file at the root of this repository.
 5#
 6# As of the Change Date specified in that file, in accordance with
 7# the Business Source License, use of this software will be governed
 8# by the Apache License, Version 2.0.
 9
10from typing import Union
11
12from materialize.zippy.debezium_capabilities import DebeziumSourceExists
13from materialize.zippy.mysql_cdc_capabilities import MySqlCdcTableExists
14from materialize.zippy.pg_cdc_capabilities import PostgresCdcTableExists
15from materialize.zippy.source_capabilities import SourceExists
16from materialize.zippy.table_capabilities import TableExists
17from materialize.zippy.watermarked_object_capabilities import WatermarkedObjectExists
18from materialize.zippy.watermarks import Watermarks
19
20WatermarkedObjectExistss = list[
21    Union[
22        TableExists,
23        SourceExists,
24        "ViewExists",
25        DebeziumSourceExists,
26        PostgresCdcTableExists,
27        MySqlCdcTableExists,
28    ]
29]
30
31
32class ViewExists(WatermarkedObjectExists):
33    """A view exists in Materialize."""
34
35    @classmethod
36    def format_str(cls) -> str:
37        return "view_{}"
38
39    def __init__(
40        self,
41        name: str,
42        inputs: list[WatermarkedObjectExists],
43        expensive_aggregates: bool | None = None,
44        has_index: bool = False,
45    ) -> None:
46        self.name = name
47        self.inputs = inputs
48        self.expensive_aggregates = expensive_aggregates
49        self.has_index = has_index
50
51    def get_watermarks(self) -> Watermarks:
52        """Calculate the intersection of the mins/maxs of the inputs. The result from the view should match the calculation."""
53
54        return Watermarks(
55            min_watermark=max([f.get_watermarks().min for f in self.inputs]),
56            max_watermark=min([f.get_watermarks().max for f in self.inputs]),
57        )
WatermarkedObjectExistss = list[typing.Union[materialize.zippy.table_capabilities.TableExists, materialize.zippy.source_capabilities.SourceExists, ForwardRef('ViewExists'), materialize.zippy.debezium_capabilities.DebeziumSourceExists, materialize.zippy.pg_cdc_capabilities.PostgresCdcTableExists, materialize.zippy.mysql_cdc_capabilities.MySqlCdcTableExists]]
class ViewExists(materialize.zippy.watermarked_object_capabilities.WatermarkedObjectExists): 
33class ViewExists(WatermarkedObjectExists):
34    """A view exists in Materialize."""
35
36    @classmethod
37    def format_str(cls) -> str:
38        return "view_{}"
39
40    def __init__(
41        self,
42        name: str,
43        inputs: list[WatermarkedObjectExists],
44        expensive_aggregates: bool | None = None,
45        has_index: bool = False,
46    ) -> None:
47        self.name = name
48        self.inputs = inputs
49        self.expensive_aggregates = expensive_aggregates
50        self.has_index = has_index
51
52    def get_watermarks(self) -> Watermarks:
53        """Calculate the intersection of the mins/maxs of the inputs. The result from the view should match the calculation."""
54
55        return Watermarks(
56            min_watermark=max([f.get_watermarks().min for f in self.inputs]),
57            max_watermark=min([f.get_watermarks().max for f in self.inputs]),
58        )

A view exists in Materialize.

ViewExists( name: str, inputs: list[materialize.zippy.watermarked_object_capabilities.WatermarkedObjectExists], expensive_aggregates: bool | None = None, has_index: bool = False) 
40    def __init__(
41        self,
42        name: str,
43        inputs: list[WatermarkedObjectExists],
44        expensive_aggregates: bool | None = None,
45        has_index: bool = False,
46    ) -> None:
47        self.name = name
48        self.inputs = inputs
49        self.expensive_aggregates = expensive_aggregates
50        self.has_index = has_index
@classmethod
def format_str(cls) -> str: 
36    @classmethod
37    def format_str(cls) -> str:
38        return "view_{}"
name
inputs
expensive_aggregates
has_index
def get_watermarks(self) -> materialize.zippy.watermarks.Watermarks: 
52    def get_watermarks(self) -> Watermarks:
53        """Calculate the intersection of the mins/maxs of the inputs. The result from the view should match the calculation."""
54
55        return Watermarks(
56            min_watermark=max([f.get_watermarks().min for f in self.inputs]),
57            max_watermark=min([f.get_watermarks().max for f in self.inputs]),
58        )

Calculate the intersection of the mins/maxs of the inputs. The result from the view should match the calculation.