data_algebra/arrow.py
def apply_to(self, b):
        """replace self input table with b"""
        if isinstance(b, data_algebra.data_ops.ViewRepresentation):
            b = DataOpArrow(b)
        assert isinstance(b, DataOpArrow)
        # check categorical arrow composition conditions
        missing = set(self.incoming_columns) - set(b.outgoing_columns)
        if len(missing) > 0:
            raise ValueError("missing required columns: " + str(missing))
        problem_production = set(self.forbidden_columns()) - set(b.forbidden_to_produce)
        if len(problem_production) > 0:
            raise ValueError(
                "did not document non-production of columns: " + str(problem_production)
            )
        excess = set(b.outgoing_columns) - set(self.incoming_columns)
        if len(excess) > 0:
            problem_excess = excess.intersection(self.forbidden_columns())
            if len(problem_excess) > 0:
                raise ValueError("forbidden incoming columns: " + str(excess))
            if self.strict:
                raise ValueError("extra incoming columns: " + str(excess))
        if (self.incoming_types is not None) and (b.outgoing_types is not None):
            for c in self.incoming_columns:
                st = self.incoming_types[c]
                xt = b.outgoing_types[c]
                if st != xt:
                    raise ValueError(
                        "column "
                        + c
                        + " self incoming type is "
                        + str(st)
                        + ", while b outgoing type is "
                        + str(xt)
                    )
        new_pipeline = self.pipeline.apply_to(
            b.pipeline, target_table_key=self.free_table_key
        )
        new_pipeline.get_tables()  # check tables are compatible
        res = DataOpArrow(
            pipeline=new_pipeline,
            free_table_key=b.free_table_key,
            forbidden_to_produce=self.forbidden_to_produce,
        )
        res.incoming_types = b.incoming_types
        res.outgoing_types = self.outgoing_types
        return res
replace self input table with b
replace self input table with b
replace self input table with b
def apply_to(self, b):
        if isinstance(b, data_algebra.data_ops.ViewRepresentation):
            b = DataOpArrow(b)
        assert isinstance(b, DataOpArrow)
        missing = set(self.incoming_columns) - set(b.outgoing_columns)
        if len(missing) > 0:
            raise ValueError("missing required columns: " + str(missing))
        problem_production = set(self.forbidden_columns()) - set(b.forbidden_to_produce)
        if len(problem_production) > 0:
            raise ValueError(
                "did not document non-production of columns: " + str(problem_production)
            )
        excess = set(b.outgoing_columns) - set(self.incoming_columns)
        if len(excess) > 0:
            problem_excess = excess.intersection(self.forbidden_columns())
            if len(problem_excess) > 0:
                raise ValueError("forbidden incoming columns: " + str(excess))
            if self.strict:
                raise ValueError("extra incoming columns: " + str(excess))
        if (self.incoming_types is not None) and (b.outgoing_types is not None):
            for c in self.incoming_columns:
                st = self.incoming_types[c]
                xt = b.outgoing_types[c]
                if st != xt:
                    raise ValueError(
                        "column "
                        + c
                        + " self incoming type is "
                        + str(st)
                        + ", while b outgoing type is "
                        + str(xt)
                    )
        new_pipeline = self.pipeline.apply_to(
            b.pipeline, target_table_key=self.free_table_key
        )
        new_pipeline.get_tables()  
        res = DataOpArrow(
            pipeline=new_pipeline,
            free_table_key=b.free_table_key,
            forbidden_to_produce=self.forbidden_to_produce,
        )
        res.incoming_types = b.incoming_types
        res.outgoing_types = self.outgoing_types
        return res
def apply_to ( self , b ) : if isinstance ( b , data_algebra . data_ops . ViewRepresentation ) : b = DataOpArrow ( b ) assert isinstance ( b , DataOpArrow ) missing = set ( self . incoming_columns ) - set ( b . outgoing_columns ) if len ( missing ) > 0 : raise ValueError ( "missing required columns: " + str ( missing ) ) problem_production = set ( self . forbidden_columns ( ) ) - set ( b . forbidden_to_produce ) if len ( problem_production ) > 0 : raise ValueError ( "did not document non-production of columns: " + str ( problem_production ) ) excess = set ( b . outgoing_columns ) - set ( self . incoming_columns ) if len ( excess ) > 0 : problem_excess = excess . intersection ( self . forbidden_columns ( ) ) if len ( problem_excess ) > 0 : raise ValueError ( "forbidden incoming columns: " + str ( excess ) ) if self . strict : raise ValueError ( "extra incoming columns: " + str ( excess ) ) if ( self . incoming_types is not None ) and ( b . outgoing_types is not None ) : for c in self . incoming_columns : st = self . incoming_types [ c ] xt = b . outgoing_types [ c ] if st != xt : raise ValueError ( "column " + c + " self incoming type is " + str ( st ) + ", while b outgoing type is " + str ( xt ) ) new_pipeline = self . pipeline . apply_to ( b . pipeline , target_table_key = self . free_table_key ) new_pipeline . get_tables ( ) res = DataOpArrow ( pipeline = new_pipeline , free_table_key = b . free_table_key , forbidden_to_produce = self . forbidden_to_produce , ) res . incoming_types = b . incoming_types res . outgoing_types = self . outgoing_types return res
replace self input table with b
replace self input table with b
"""replace self input table with b""" # check categorical arrow composition conditions # check tables are compatible
self
b