pathway package

Pathway main module

Typical use:

 import pathway as pw

class pathway.AsofJoinResult(side_data, mode, defaults)

Result of an ASOF join of two tables

Example:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
         | K | val |  t
     1   | 0 | 1   |  1
     2   | 0 | 2   |  4
     3   | 0 | 3   |  5
     4   | 0 | 4   |  6
     5   | 0 | 5   |  7
     6   | 0 | 6   |  11
     7   | 0 | 7   |  12
     8   | 1 | 8   |  5
     9   | 1 | 9   |  7
 '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
             | K | val | t
     21   | 1 | 7  | 2
     22   | 1 | 3  | 8
     23   | 0 | 0  | 2
     24   | 0 | 6  | 3
     25   | 0 | 2  | 7
     26   | 0 | 3  | 8
     27   | 0 | 9  | 9
     28   | 0 | 7  | 13
     29   | 0 | 4  | 14
     '''
 )
 res = t1.asof_join(
     t2,
     t1.t,
     t2.t,
     t1.K == t2.K,
     mode=pw.JoinMode.LEFT,
     defaults={t2.val: -1},
 ).select(
     pw.this.shard_key,
     pw.this.t,
     val_left=t1.val,
     val_right=t2.val,
     sum=t1.val + t2.val,
 )
 pw.debug.compute_and_print(res, include_id=False)

shard_key | t  | val_left | val_right | sum
0         | 1  | 1        | -1        | 0
0         | 4  | 2        | 6         | 8
0         | 5  | 3        | 6         | 9
0         | 6  | 4        | 6         | 10
0         | 7  | 5        | 6         | 11
0         | 11 | 6        | 9         | 15
0         | 12 | 7        | 9         | 16
1         | 5  | 8        | 7         | 15
1         | 7  | 9        | 7         | 16

class pathway.AsyncTransformer(input_table)

Allows to perform async transformations on a table.

invoke() will be called asynchronously for each row of an input_table.

Output table can be acccesed via result.

Example:

 import pathway as pw
 import asyncio
 class OutputSchema(pw.Schema):
    ret: int
 class AsyncIncrementTransformer(pw.AsyncTransformer, output_schema=OutputSchema):
     async def invoke(self, value) -> Dict[str, Any]:
         await asyncio.sleep(0.1)
         return {"ret": value + 1 }
 input = pw.debug.parse_to_table('''
   | value
 1 | 42
 2 | 44
 ''')
 result = AsyncIncrementTransformer(input_table=input).result
 pw.debug.compute_and_print(result, include_id=False)

ret
42
43

close()

Called once at the end. Proper place for cleanup.

• Return type
  None

abstract async invoke(*args, **kwargs)

Called for every row of input_table. The arguments will correspond to the columns in the input table.

Should return dict of values matching output_schema.

• Return type
  Dictstr, Any

open()

Called before actual work. Suitable for one time setup.

• Return type
  None

property result(: Tabl )

Resulting table.

with_options(capacity=None, retry_strategy=None)

Sets async options.

• Parameters
  • capacity (Optionalint) – maximum number of concurrent operations.
  • retry_strategy (OptionalAsyncRetryStrategy) – defines how failures will be handled.
• Return type
  AsyncTransformer
• Returns
  self

class pathway.ClassArg(ref: RowReference, ptr: Pointer)

Base class to inherit from when writing inner classes for class transformers.

Example:

 import pathway as pw
 @pw.transformer
 class simple_transformer:
     class table(pw.ClassArg):
         arg = pw.input_attribute()
         @pw.output_attribute
         def ret(self) -> int:
             return self.arg + 1
 t1 = pw.debug.parse_to_table('''
 age
 10
 9
 8
 7
 ''')
 t2 = simple_transformer(table=t1.select(arg=t1.age)).table
 pw.debug.compute_and_print(t1 + t2, include_id=False)

age | ret
7   | 8
8   | 9
9   | 10
10  | 11

pointer_from(*args, optional=False)

Pseudo-random hash of its argument. Produces pointer types. Applied value-wise.

• Return type
  BasePointer

class pathway.ColumnReference(*, column, table, name)

Reference to the column.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
    age  owner  pet
 1   10  Alice  dog
 2    9    Bob  dog
 3    8  Alice  cat
 4    7    Bob  dog''')
 isinstance(t1.age, pw.ColumnReference)

True

 isinstance(t1["owner"], pw.ColumnReference)

True

property name()

Name of the referred column.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
    age  owner  pet
 1   10  Alice  dog
 2    9    Bob  dog
 3    8  Alice  cat
 4    7    Bob  dog''')
 t1.age.name

'age'

property table()

Table where the referred column belongs to.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
    age  owner  pet
 1   10  Alice  dog
 2    9    Bob  dog
 3    8  Alice  cat
 4    7    Bob  dog''')
 t1.age.table is t1

True

class pathway.FilteredJoinResult(join_result, filtering)

filter(filter_expression)

Filters rows, keeping the ones satisfying the predicate.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
    age  owner  pet
 1   10  Alice    1
 2    9    Bob    1
 3    8  Alice    2
 ''')
 t2 = pw.debug.parse_to_table('''
     age  owner  pet size
 11   10  Alice    3    M
 12    9    Bob    1    L
 13    8    Tom    1   XL
 ''')
 result = t1.join(t2).filter(t1.owner == t2.owner).select(t1.age, t2.size)   # noqa: E501
 pw.debug.compute_and_print(result, include_id=False)

age | size
8   | M
9   | L
10  | M

• Return type
  FilteredJoinResult

groupby(*args, id=None)

Groups join result by columns from args.

NOTE: Usually followed by .reduce() that aggregates the result and returns a table.

• Parameters
  • args (ColumnReference) – columns to group by.
  • id (OptionalColumnReference) – if provided, is the column used to set id’s of the rows of the result
• Returns
  Groupby object.
• Return type
  GroupedJoinResult

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
    cost  owner  pet
 1   100  Alice    1
 2    90    Bob    1
 3    80  Alice    2
 ''')
 t2 = pw.debug.parse_to_table('''
     cost  owner  pet size
 11   100  Alice    3    M
 12    90    Bob    1    L
 13    80    Tom    1   XL
 ''')
 result = (t1.join(t2, t1.owner==t2.owner).groupby(pw.this.owner)
     .reduce(pw.this.owner, pairs = pw.reducers.count()))
 pw.debug.compute_and_print(result, include_id=False)

owner | pairs
Alice | 2
Bob   | 1

join(other, *on, id=None, how=JoinMode.INNER)

Join self with other using the given join expression.

• Parameters
  • other (Joinable) – the right side of the join.
  • on (ColumnBinaryOpExpression) – a list of column expressions. Each must have == as the top level operation and be of the form LHS: ColumnReference == RHS: ColumnReference.
  • id (OptionalColumnReference) – optional argument for id of result, can be only self.id or other.id
  • how (JoinMode) – by default, inner join is performed. Possible values are JoinMode.{INNER,LEFT,RIGHT,OUTER} correspond to inner, left, right and outer join respectively.
• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age  | owner  | pet
  10  | Alice  | 1
   9  | Bob    | 1
   8  | Alice  | 2
 ''')
 t2 = pw.debug.parse_to_table('''
 age  | owner  | pet | size
  10  | Alice  | 3   | M
  9   | Bob    | 1   | L
  8   | Tom    | 1   | XL
 ''')
 t3 = t1.join(t2, t1.pet == t2.pet, t1.owner == t2.owner, how=pw.JoinMode.INNER).select(age=t1.age, owner_name=t2.owner, size=t2.size)  # noqa: E501
 pw.debug.compute_and_print(t3, include_id = False)

age | owner_name | size
9   | Bob        | L

join_inner(other, *on, id=None)

Inner-joins two tables or join results.

• Parameters
  • other (Joinable) – the right side of the join.
  • on (ColumnBinaryOpExpression) – a list of column expressions. Each must have == as the top level operation and be of the form LHS: ColumnReference == RHS: ColumnReference.
  • id (OptionalColumnReference) – optional argument for id of result, can be only self.id or other.id
• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age  | owner  | pet
  10  | Alice  | 1
   9  | Bob    | 1
   8  | Alice  | 2
 ''')
 t2 = pw.debug.parse_to_table('''
 age  | owner  | pet | size
  10  | Alice  | 3   | M
  9   | Bob    | 1   | L
  8   | Tom    | 1   | XL
 ''')
 t3 = t1.join(t2, t1.pet == t2.pet, t1.owner == t2.owner, how=pw.JoinMode.INNER).select(age=t1.age, owner_name=t2.owner, size=t2.size)  # noqa: E501
 pw.debug.compute_and_print(t3, include_id = False)

age | owner_name | size
9   | Bob        | L

join_left(other, *on, id=None)

Left-joins two tables or join results.

• Parameters
  • other (Joinable) – Table or join result.
  • *on – Columns to join, syntax self.col1 == other.col2
  • id (OptionalColumnReference) – optional id column of the result

Remarks: args cannot contain id column from either of tables, as the result table has id column with auto-generated ids; it can be selected by assigning it to a column with defined name (passed in kwargs)

Behavior:

• for rows from the left side that were not matched with the right side, missing values on the right are replaced with None
• rows from the right side that were not matched with the left side are skipped
• for rows that were matched the behavior is the same as that of an inner join.

• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
         | A  | B
       1 | 11 | 111
       2 | 12 | 112
       3 | 13 | 113
       4 | 13 | 114
     '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
         | C  | D
       1 | 11 | 211
       2 | 12 | 212
       3 | 14 | 213
       4 | 14 | 214
     '''
 )
 pw.debug.compute_and_print(t1.join_left(t2, t1.A == t2.C
 ).select(t1.A, t2_C=t2.C, S=pw.require(t1.B + t2.D,t2.id)),
 include_id=False)

A  | t2_C | S
11 | 11   | 322
12 | 12   | 324
13 |      |
13 |      |

join_outer(other, *on, id=None)

Outer-joins two tables or join results.

• Parameters
  • other (Joinable) – Table or join result.
  • *on – Columns to join, syntax self.col1 == other.col2
  • id (OptionalColumnReference) – optional id column of the result

Remarks: args cannot contain id column from either of tables, as the result table has id column with auto-generated ids; it can be selected by assigning it to a column with defined name (passed in kwargs)

Behavior:

• for rows from the left side that were not matched with the right side, missing values on the right are replaced with None
• for rows from the right side that were not matched with the left side, missing values on the left are replaced with None
• for rows that were matched the behavior is the same as that of an inner join.

• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
         | A  | B
       1 | 11 | 111
       2 | 12 | 112
       3 | 13 | 113
       4 | 13 | 114
     '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
         | C  | D
       1 | 11 | 211
       2 | 12 | 212
       3 | 14 | 213
       4 | 14 | 214
     '''
 )
 pw.debug.compute_and_print(t1.join_outer(t2, t1.A == t2.C
 ).select(t1.A, t2_C=t2.C, S=pw.require(t1.B + t2.D,t1.id,t2.id)),
 include_id=False)

A  | t2_C | S
   | 14   |
   | 14   |
11 | 11   | 322
12 | 12   | 324
13 |      |
13 |      |

join_right(other, *on, id=None)

Outer-joins two tables or join results.

• Parameters
  • other (Joinable) – Table or join result.
  • *on – Columns to join, syntax self.col1 == other.col2
  • id (OptionalColumnReference) – optional id column of the result

Remarks: args cannot contain id column from either of tables, as the result table has id column with auto-generated ids; it can be selected by assigning it to a column with defined name (passed in kwargs)

Behavior:

• rows from the left side that were not matched with the right side are skipped
• for rows from the right side that were not matched with the left side, missing values on the left are replaced with None
• for rows that were matched the behavior is the same as that of an inner join.

• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
         | A  | B
       1 | 11 | 111
       2 | 12 | 112
       3 | 13 | 113
       4 | 13 | 114
     '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
         | C  | D
       1 | 11 | 211
       2 | 12 | 212
       3 | 14 | 213
       4 | 14 | 214
     '''
 )
 pw.debug.compute_and_print(t1.join_right(t2, t1.A == t2.C
 ).select(t1.A, t2_C=t2.C, S=pw.require(pw.coalesce(t1.B,0) + t2.D,t1.id)),
 include_id=False)

A  | t2_C | S
   | 14   |
   | 14   |
11 | 11   | 322
12 | 12   | 324

• Returns
  OuterJoinResult object

promise_universe_is_equal_to(other)

Asserts to Pathway that an universe of self is a subset of universe of each of the others.

Semantics: Used in situations where Pathway cannot deduce one universe being a subset of another.

Returns: None

Note: The assertion works in place.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | pet
 1 | Dog
 7 | Cat
 ''')
 t2 = pw.debug.parse_to_table('''
   | age
 1 | 10
 7 | 3
 ''')
 t1 = t1.promise_universe_is_equal_to(t2)
 t3 = t1 + t2
 pw.debug.compute_and_print(t3, include_id=False)

pet | age
Cat | 3
Dog | 10

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

promise_universe_is_subset_of(other)

Asserts to Pathway that an universe of self is a subset of universe of each of the other.

Semantics: Used in situations where Pathway cannot deduce one universe being a subset of another.

Returns: self

Note: The assertion works in place.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 1
 2 | 9   | Bob   | 1
 3 | 8   | Alice | 2
 ''')
 t2 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 30
 ''').promise_universe_is_subset_of(t1)
 t3 = t1 << t2
 pw.debug.compute_and_print(t3, include_id=False)

age | owner | pet
8   | Alice | 2
9   | Bob   | 1
10  | Alice | 30

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

promise_universes_are_disjoint(other)

Asserts to Pathway that an universe of self is disjoint from universe of other.

Semantics: Used in situations where Pathway cannot deduce universes are disjoint.

Returns: self

Note: The assertion works in place.

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 1
 2 | 9   | Bob   | 1
 3 | 8   | Alice | 2
 ''')
 t2 = pw.debug.parse_to_table('''
    | age | owner | pet
 11 | 11  | Alice | 30
 12 | 12  | Tom   | 40
 ''').promise_universes_are_disjoint(t1)
 t3 = t1.concat(t2)
 pw.debug.compute_and_print(t3, include_id=False)

age | owner | pet
8   | Alice | 2
9   | Bob   | 1
10  | Alice | 1
11  | Alice | 30
12  | Tom   | 40

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

reduce(*args, **kwargs)

Reduce a join result to a single row.

Equivalent to self.groupby().reduce(*args, **kwargs).

• Parameters
  • args (ColumnReference) – reducer to reduce the table with
  • kwargs (ColumnExpression) – reducer to reduce the table with. Its key is the new name of a column.
• Returns
  Reduced table.
• Return type
  Table

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
    cost  owner  pet
 1   100  Alice    1
 2    90    Bob    1
 3    80  Alice    2
 ''')
 t2 = pw.debug.parse_to_table('''
     cost  owner  pet size
 11   100  Alice    3    M
 12    90    Bob    1    L
 13    80    Tom    1   XL
 ''')
 result = t1.join(t2, t1.owner==t2.owner).reduce(total_pairs = pw.reducers.count())
 pw.debug.compute_and_print(result, include_id=False)

total_pairs
3

select(*args, **kwargs)

Computes result of a join.

• Parameters
  • args (ColumnReference) – Column references.
  • kwargs (Any) – Column expressions with their new assigned names.
• Returns
  Created table.
• Return type
  Table

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age  | owner  | pet
  10  | Alice  | 1
   9  | Bob    | 1
   8  | Alice  | 2
 ''')
 t2 = pw.debug.parse_to_table('''
 age  | owner  | pet | size
  10  | Alice  | 3   | M
  9   | Bob    | 1   | L
  8   | Tom    | 1   | XL
 ''')
 t3 = t1.join(t2, t1.pet == t2.pet, t1.owner == t2.owner).select(age=t1.age, owner_name=t2.owner, size=t2.size)   # noqa: E501
 pw.debug.compute_and_print(t3, include_id=False)

age | owner_name | size
9   | Bob        | L

class pathway.GroupedJoinResult(*, join_result, args, id)

promise_universe_is_equal_to(other)

Asserts to Pathway that an universe of self is a subset of universe of each of the others.

Semantics: Used in situations where Pathway cannot deduce one universe being a subset of another.

Returns: None

Note: The assertion works in place.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | pet
 1 | Dog
 7 | Cat
 ''')
 t2 = pw.debug.parse_to_table('''
   | age
 1 | 10
 7 | 3
 ''')
 t1 = t1.promise_universe_is_equal_to(t2)
 t3 = t1 + t2
 pw.debug.compute_and_print(t3, include_id=False)

pet | age
Cat | 3
Dog | 10

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

promise_universe_is_subset_of(other)

Asserts to Pathway that an universe of self is a subset of universe of each of the other.

Semantics: Used in situations where Pathway cannot deduce one universe being a subset of another.

Returns: self

Note: The assertion works in place.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 1
 2 | 9   | Bob   | 1
 3 | 8   | Alice | 2
 ''')
 t2 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 30
 ''').promise_universe_is_subset_of(t1)
 t3 = t1 << t2
 pw.debug.compute_and_print(t3, include_id=False)

age | owner | pet
8   | Alice | 2
9   | Bob   | 1
10  | Alice | 30

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

promise_universes_are_disjoint(other)

Asserts to Pathway that an universe of self is disjoint from universe of other.

Semantics: Used in situations where Pathway cannot deduce universes are disjoint.

Returns: self

Note: The assertion works in place.

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 1
 2 | 9   | Bob   | 1
 3 | 8   | Alice | 2
 ''')
 t2 = pw.debug.parse_to_table('''
    | age | owner | pet
 11 | 11  | Alice | 30
 12 | 12  | Tom   | 40
 ''').promise_universes_are_disjoint(t1)
 t3 = t1.concat(t2)
 pw.debug.compute_and_print(t3, include_id=False)

age | owner | pet
8   | Alice | 2
9   | Bob   | 1
10  | Alice | 1
11  | Alice | 30
12  | Tom   | 40

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

reduce(*args, **kwargs)

Reduces grouped join result to table.

• Returns
  Created table.
• Return type
  Table

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
    cost  owner  pet
 1   100  Alice    1
 2    90    Bob    1
 3    80  Alice    2
 ''')
 t2 = pw.debug.parse_to_table('''
     cost  owner  pet size
 11   100  Alice    3    M
 12    90    Bob    1    L
 13    80    Tom    1   XL
 ''')
 result = (t1.join(t2, t1.owner==t2.owner).groupby(pw.this.owner)
     .reduce(pw.this.owner, pairs = pw.reducers.count()))
 pw.debug.compute_and_print(result, include_id=False)

owner | pairs
Alice | 2
Bob   | 1

class pathway.GroupedTable(table, grouping_columns, set_id=False)

Result of a groupby operation on a Table.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age | owner | pet
 10  | Alice | dog
 9   | Bob   | dog
 8   | Alice | cat
 7   | Bob   | dog
 ''')
 t2 = t1.groupby(t1.pet, t1.owner)
 isinstance(t2, pw.GroupedTable)

True

promise_universe_is_equal_to(other)

Asserts to Pathway that an universe of self is a subset of universe of each of the others.

Semantics: Used in situations where Pathway cannot deduce one universe being a subset of another.

Returns: None

Note: The assertion works in place.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | pet
 1 | Dog
 7 | Cat
 ''')
 t2 = pw.debug.parse_to_table('''
   | age
 1 | 10
 7 | 3
 ''')
 t1 = t1.promise_universe_is_equal_to(t2)
 t3 = t1 + t2
 pw.debug.compute_and_print(t3, include_id=False)

pet | age
Cat | 3
Dog | 10

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

promise_universe_is_subset_of(other)

Asserts to Pathway that an universe of self is a subset of universe of each of the other.

Semantics: Used in situations where Pathway cannot deduce one universe being a subset of another.

Returns: self

Note: The assertion works in place.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 1
 2 | 9   | Bob   | 1
 3 | 8   | Alice | 2
 ''')
 t2 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 30
 ''').promise_universe_is_subset_of(t1)
 t3 = t1 << t2
 pw.debug.compute_and_print(t3, include_id=False)

age | owner | pet
8   | Alice | 2
9   | Bob   | 1
10  | Alice | 30

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

promise_universes_are_disjoint(other)

Asserts to Pathway that an universe of self is disjoint from universe of other.

Semantics: Used in situations where Pathway cannot deduce universes are disjoint.

Returns: self

Note: The assertion works in place.

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 1
 2 | 9   | Bob   | 1
 3 | 8   | Alice | 2
 ''')
 t2 = pw.debug.parse_to_table('''
    | age | owner | pet
 11 | 11  | Alice | 30
 12 | 12  | Tom   | 40
 ''').promise_universes_are_disjoint(t1)
 t3 = t1.concat(t2)
 pw.debug.compute_and_print(t3, include_id=False)

age | owner | pet
8   | Alice | 2
9   | Bob   | 1
10  | Alice | 1
11  | Alice | 30
12  | Tom   | 40

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

reduce(*args, **kwargs)

Reduces grouped table to a table.

• Parameters
  • args (ColumnReference) – Column references.
  • kwargs (ColumnExpression) – Column expressions with their new assigned names.
• Returns
  Created table.
• Return type
  Table

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age | owner | pet
 10  | Alice | dog
 9   | Bob   | dog
 8   | Alice | cat
 7   | Bob   | dog
 ''')
 t2 = t1.groupby(t1.pet, t1.owner).reduce(t1.owner, t1.pet, ageagg=pw.reducers.sum(t1.age))
 pw.debug.compute_and_print(t2, include_id=False)

owner | pet | ageagg
Alice | cat | 8
Alice | dog | 10
Bob   | dog | 16

class pathway.IntervalJoinResult(left_bucketed, right_bucketed, earlier_part_filtered, later_part_filtered, table_substitution, mode)

Result of an interval join between tables.

Example:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
     | t
   1 | 3
   2 | 4
   3 | 5
   4 | 11
 '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
     | t
   1 | 0
   2 | 1
   3 | 4
   4 | 7
 '''
 )
 join_result = t1.interval_join(t2, t1.t, t2.t, -2, 1)
 isinstance(join_result, pw.IntervalJoinResult)

True

 pw.debug.compute_and_print(
     join_result.select(left_t=t1.t, right_t=t2.t), include_id=False
 )

left_t | right_t
3      | 1
3      | 4
4      | 4
5      | 4

select(*args, **kwargs)

Computes a result of an interval join.

• Parameters
  • args (ColumnReference) – Column references.
  • kwargs (Any) – Column expressions with their new assigned names.
• Returns
  Created table.
• Return type
  Table

Example:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
     | a | t
   1 | 1 | 3
   2 | 1 | 4
   3 | 1 | 5
   4 | 1 | 11
   5 | 2 | 2
   6 | 2 | 3
   7 | 3 | 4
 '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
     | b | t
   1 | 1 | 0
   2 | 1 | 1
   3 | 1 | 4
   4 | 1 | 7
   5 | 2 | 0
   6 | 2 | 2
   7 | 4 | 2
 '''
 )
 t3 = t1.interval_join(t2, t1.t, t2.t, -2, 1, t1.a == t2.b).select(
     t1.a, left_t=t1.t, right_t=t2.t
 )
 pw.debug.compute_and_print(t3, include_id=False)

a | left_t | right_t
1 | 3      | 1
1 | 3      | 4
1 | 4      | 4
1 | 5      | 4
2 | 2      | 0
2 | 2      | 2
2 | 3      | 2

class pathway.JoinMode(value)

An enumeration.

class pathway.JoinResult(_universe, _context, _left_table, _right_table, _original_left, _original_right, _substitution, _chained_join_desugaring, _joined_on_names, _join_mode)

Result of a join between tables.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
    age  owner  pet
 1   10  Alice    1
 2    9    Bob    1
 3    8  Alice    2
 ''')
 t2 = pw.debug.parse_to_table('''
     age  owner  pet size
 11   10  Alice    3    M
 12    9    Bob    1    L
 13    8    Tom    1   XL
 ''')
 joinresult= t1.join(t2, t1.pet == t2.pet, t1.owner == t2.owner)   # noqa: E501
 isinstance(joinresult, pw.JoinResult)

True

 pw.debug.compute_and_print(joinresult.select(t1.age, t2.size), include_id=False)

age | size
9   | L

filter(filter_expression)

Filters rows, keeping the ones satisfying the predicate.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
    age  owner  pet
 1   10  Alice    1
 2    9    Bob    1
 3    8  Alice    2
 ''')
 t2 = pw.debug.parse_to_table('''
     age  owner  pet size
 11   10  Alice    3    M
 12    9    Bob    1    L
 13    8    Tom    1   XL
 ''')
 result = t1.join(t2).filter(t1.owner == t2.owner).select(t1.age, t2.size)   # noqa: E501
 pw.debug.compute_and_print(result, include_id=False)

age | size
8   | M
9   | L
10  | M

• Return type
  FilteredJoinResult

groupby(*args, id=None)

Groups join result by columns from args.

NOTE: Usually followed by .reduce() that aggregates the result and returns a table.

• Parameters
  • args (ColumnReference) – columns to group by.
  • id (OptionalColumnReference) – if provided, is the column used to set id’s of the rows of the result
• Returns
  Groupby object.
• Return type
  GroupedJoinResult

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
    cost  owner  pet
 1   100  Alice    1
 2    90    Bob    1
 3    80  Alice    2
 ''')
 t2 = pw.debug.parse_to_table('''
     cost  owner  pet size
 11   100  Alice    3    M
 12    90    Bob    1    L
 13    80    Tom    1   XL
 ''')
 result = (t1.join(t2, t1.owner==t2.owner).groupby(pw.this.owner)
     .reduce(pw.this.owner, pairs = pw.reducers.count()))
 pw.debug.compute_and_print(result, include_id=False)

owner | pairs
Alice | 2
Bob   | 1

join(other, *on, id=None, how=JoinMode.INNER)

Join self with other using the given join expression.

• Parameters
  • other (Joinable) – the right side of the join.
  • on (ColumnBinaryOpExpression) – a list of column expressions. Each must have == as the top level operation and be of the form LHS: ColumnReference == RHS: ColumnReference.
  • id (OptionalColumnReference) – optional argument for id of result, can be only self.id or other.id
  • how (JoinMode) – by default, inner join is performed. Possible values are JoinMode.{INNER,LEFT,RIGHT,OUTER} correspond to inner, left, right and outer join respectively.
• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age  | owner  | pet
  10  | Alice  | 1
   9  | Bob    | 1
   8  | Alice  | 2
 ''')
 t2 = pw.debug.parse_to_table('''
 age  | owner  | pet | size
  10  | Alice  | 3   | M
  9   | Bob    | 1   | L
  8   | Tom    | 1   | XL
 ''')
 t3 = t1.join(t2, t1.pet == t2.pet, t1.owner == t2.owner, how=pw.JoinMode.INNER).select(age=t1.age, owner_name=t2.owner, size=t2.size)  # noqa: E501
 pw.debug.compute_and_print(t3, include_id = False)

age | owner_name | size
9   | Bob        | L

join_inner(other, *on, id=None)

Inner-joins two tables or join results.

• Parameters
  • other (Joinable) – the right side of the join.
  • on (ColumnBinaryOpExpression) – a list of column expressions. Each must have == as the top level operation and be of the form LHS: ColumnReference == RHS: ColumnReference.
  • id (OptionalColumnReference) – optional argument for id of result, can be only self.id or other.id
• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age  | owner  | pet
  10  | Alice  | 1
   9  | Bob    | 1
   8  | Alice  | 2
 ''')
 t2 = pw.debug.parse_to_table('''
 age  | owner  | pet | size
  10  | Alice  | 3   | M
  9   | Bob    | 1   | L
  8   | Tom    | 1   | XL
 ''')
 t3 = t1.join(t2, t1.pet == t2.pet, t1.owner == t2.owner, how=pw.JoinMode.INNER).select(age=t1.age, owner_name=t2.owner, size=t2.size)  # noqa: E501
 pw.debug.compute_and_print(t3, include_id = False)

age | owner_name | size
9   | Bob        | L

join_left(other, *on, id=None)

Left-joins two tables or join results.

• Parameters
  • other (Joinable) – Table or join result.
  • *on – Columns to join, syntax self.col1 == other.col2
  • id (OptionalColumnReference) – optional id column of the result

Remarks: args cannot contain id column from either of tables, as the result table has id column with auto-generated ids; it can be selected by assigning it to a column with defined name (passed in kwargs)

Behavior:

• for rows from the left side that were not matched with the right side, missing values on the right are replaced with None
• rows from the right side that were not matched with the left side are skipped
• for rows that were matched the behavior is the same as that of an inner join.

• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
         | A  | B
       1 | 11 | 111
       2 | 12 | 112
       3 | 13 | 113
       4 | 13 | 114
     '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
         | C  | D
       1 | 11 | 211
       2 | 12 | 212
       3 | 14 | 213
       4 | 14 | 214
     '''
 )
 pw.debug.compute_and_print(t1.join_left(t2, t1.A == t2.C
 ).select(t1.A, t2_C=t2.C, S=pw.require(t1.B + t2.D,t2.id)),
 include_id=False)

A  | t2_C | S
11 | 11   | 322
12 | 12   | 324
13 |      |
13 |      |

join_outer(other, *on, id=None)

Outer-joins two tables or join results.

• Parameters
  • other (Joinable) – Table or join result.
  • *on – Columns to join, syntax self.col1 == other.col2
  • id (OptionalColumnReference) – optional id column of the result

Remarks: args cannot contain id column from either of tables, as the result table has id column with auto-generated ids; it can be selected by assigning it to a column with defined name (passed in kwargs)

Behavior:

• for rows from the left side that were not matched with the right side, missing values on the right are replaced with None
• for rows from the right side that were not matched with the left side, missing values on the left are replaced with None
• for rows that were matched the behavior is the same as that of an inner join.

• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
         | A  | B
       1 | 11 | 111
       2 | 12 | 112
       3 | 13 | 113
       4 | 13 | 114
     '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
         | C  | D
       1 | 11 | 211
       2 | 12 | 212
       3 | 14 | 213
       4 | 14 | 214
     '''
 )
 pw.debug.compute_and_print(t1.join_outer(t2, t1.A == t2.C
 ).select(t1.A, t2_C=t2.C, S=pw.require(t1.B + t2.D,t1.id,t2.id)),
 include_id=False)

A  | t2_C | S
   | 14   |
   | 14   |
11 | 11   | 322
12 | 12   | 324
13 |      |
13 |      |

join_right(other, *on, id=None)

Outer-joins two tables or join results.

• Parameters
  • other (Joinable) – Table or join result.
  • *on – Columns to join, syntax self.col1 == other.col2
  • id (OptionalColumnReference) – optional id column of the result

Remarks: args cannot contain id column from either of tables, as the result table has id column with auto-generated ids; it can be selected by assigning it to a column with defined name (passed in kwargs)

Behavior:

• rows from the left side that were not matched with the right side are skipped
• for rows from the right side that were not matched with the left side, missing values on the left are replaced with None
• for rows that were matched the behavior is the same as that of an inner join.

• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
         | A  | B
       1 | 11 | 111
       2 | 12 | 112
       3 | 13 | 113
       4 | 13 | 114
     '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
         | C  | D
       1 | 11 | 211
       2 | 12 | 212
       3 | 14 | 213
       4 | 14 | 214
     '''
 )
 pw.debug.compute_and_print(t1.join_right(t2, t1.A == t2.C
 ).select(t1.A, t2_C=t2.C, S=pw.require(pw.coalesce(t1.B,0) + t2.D,t1.id)),
 include_id=False)

A  | t2_C | S
   | 14   |
   | 14   |
11 | 11   | 322
12 | 12   | 324

• Returns
  OuterJoinResult object

promise_universe_is_equal_to(other)

Asserts to Pathway that an universe of self is a subset of universe of each of the others.

Semantics: Used in situations where Pathway cannot deduce one universe being a subset of another.

Returns: None

Note: The assertion works in place.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | pet
 1 | Dog
 7 | Cat
 ''')
 t2 = pw.debug.parse_to_table('''
   | age
 1 | 10
 7 | 3
 ''')
 t1 = t1.promise_universe_is_equal_to(t2)
 t3 = t1 + t2
 pw.debug.compute_and_print(t3, include_id=False)

pet | age
Cat | 3
Dog | 10

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

promise_universe_is_subset_of(other)

Asserts to Pathway that an universe of self is a subset of universe of each of the other.

Semantics: Used in situations where Pathway cannot deduce one universe being a subset of another.

Returns: self

Note: The assertion works in place.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 1
 2 | 9   | Bob   | 1
 3 | 8   | Alice | 2
 ''')
 t2 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 30
 ''').promise_universe_is_subset_of(t1)
 t3 = t1 << t2
 pw.debug.compute_and_print(t3, include_id=False)

age | owner | pet
8   | Alice | 2
9   | Bob   | 1
10  | Alice | 30

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

promise_universes_are_disjoint(other)

Asserts to Pathway that an universe of self is disjoint from universe of other.

Semantics: Used in situations where Pathway cannot deduce universes are disjoint.

Returns: self

Note: The assertion works in place.

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 1
 2 | 9   | Bob   | 1
 3 | 8   | Alice | 2
 ''')
 t2 = pw.debug.parse_to_table('''
    | age | owner | pet
 11 | 11  | Alice | 30
 12 | 12  | Tom   | 40
 ''').promise_universes_are_disjoint(t1)
 t3 = t1.concat(t2)
 pw.debug.compute_and_print(t3, include_id=False)

age | owner | pet
8   | Alice | 2
9   | Bob   | 1
10  | Alice | 1
11  | Alice | 30
12  | Tom   | 40

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

reduce(*args, **kwargs)

Reduce a join result to a single row.

Equivalent to self.groupby().reduce(*args, **kwargs).

• Parameters
  • args (ColumnReference) – reducer to reduce the table with
  • kwargs (ColumnExpression) – reducer to reduce the table with. Its key is the new name of a column.
• Returns
  Reduced table.
• Return type
  Table

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
    cost  owner  pet
 1   100  Alice    1
 2    90    Bob    1
 3    80  Alice    2
 ''')
 t2 = pw.debug.parse_to_table('''
     cost  owner  pet size
 11   100  Alice    3    M
 12    90    Bob    1    L
 13    80    Tom    1   XL
 ''')
 result = t1.join(t2, t1.owner==t2.owner).reduce(total_pairs = pw.reducers.count())
 pw.debug.compute_and_print(result, include_id=False)

total_pairs
3

select(*args, **kwargs)

Computes result of a join.

• Parameters
  • args (ColumnReference) – Column references.
  • kwargs (Any) – Column expressions with their new assigned names.
• Returns
  Created table.
• Return type
  Table

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age  | owner  | pet
  10  | Alice  | 1
   9  | Bob    | 1
   8  | Alice  | 2
 ''')
 t2 = pw.debug.parse_to_table('''
 age  | owner  | pet | size
  10  | Alice  | 3   | M
  9   | Bob    | 1   | L
  8   | Tom    | 1   | XL
 ''')
 t3 = t1.join(t2, t1.pet == t2.pet, t1.owner == t2.owner).select(age=t1.age, owner_name=t2.owner, size=t2.size)   # noqa: E501
 pw.debug.compute_and_print(t3, include_id=False)

age | owner_name | size
9   | Bob        | L

class pathway.Joinable(universe)

join(other, *on, id=None, how=JoinMode.INNER)

Join self with other using the given join expression.

• Parameters
  • other (Joinable) – the right side of the join.
  • on (ColumnBinaryOpExpression) – a list of column expressions. Each must have == as the top level operation and be of the form LHS: ColumnReference == RHS: ColumnReference.
  • id (OptionalColumnReference) – optional argument for id of result, can be only self.id or other.id
  • how (JoinMode) – by default, inner join is performed. Possible values are JoinMode.{INNER,LEFT,RIGHT,OUTER} correspond to inner, left, right and outer join respectively.
• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age  | owner  | pet
  10  | Alice  | 1
   9  | Bob    | 1
   8  | Alice  | 2
 ''')
 t2 = pw.debug.parse_to_table('''
 age  | owner  | pet | size
  10  | Alice  | 3   | M
  9   | Bob    | 1   | L
  8   | Tom    | 1   | XL
 ''')
 t3 = t1.join(t2, t1.pet == t2.pet, t1.owner == t2.owner, how=pw.JoinMode.INNER).select(age=t1.age, owner_name=t2.owner, size=t2.size)  # noqa: E501
 pw.debug.compute_and_print(t3, include_id = False)

age | owner_name | size
9   | Bob        | L

join_inner(other, *on, id=None)

Inner-joins two tables or join results.

• Parameters
  • other (Joinable) – the right side of the join.
  • on (ColumnBinaryOpExpression) – a list of column expressions. Each must have == as the top level operation and be of the form LHS: ColumnReference == RHS: ColumnReference.
  • id (OptionalColumnReference) – optional argument for id of result, can be only self.id or other.id
• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age  | owner  | pet
  10  | Alice  | 1
   9  | Bob    | 1
   8  | Alice  | 2
 ''')
 t2 = pw.debug.parse_to_table('''
 age  | owner  | pet | size
  10  | Alice  | 3   | M
  9   | Bob    | 1   | L
  8   | Tom    | 1   | XL
 ''')
 t3 = t1.join(t2, t1.pet == t2.pet, t1.owner == t2.owner, how=pw.JoinMode.INNER).select(age=t1.age, owner_name=t2.owner, size=t2.size)  # noqa: E501
 pw.debug.compute_and_print(t3, include_id = False)

age | owner_name | size
9   | Bob        | L

join_left(other, *on, id=None)

Left-joins two tables or join results.

• Parameters
  • other (Joinable) – Table or join result.
  • *on – Columns to join, syntax self.col1 == other.col2
  • id (OptionalColumnReference) – optional id column of the result

Remarks: args cannot contain id column from either of tables, as the result table has id column with auto-generated ids; it can be selected by assigning it to a column with defined name (passed in kwargs)

Behavior:

• for rows from the left side that were not matched with the right side, missing values on the right are replaced with None
• rows from the right side that were not matched with the left side are skipped
• for rows that were matched the behavior is the same as that of an inner join.

• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
         | A  | B
       1 | 11 | 111
       2 | 12 | 112
       3 | 13 | 113
       4 | 13 | 114
     '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
         | C  | D
       1 | 11 | 211
       2 | 12 | 212
       3 | 14 | 213
       4 | 14 | 214
     '''
 )
 pw.debug.compute_and_print(t1.join_left(t2, t1.A == t2.C
 ).select(t1.A, t2_C=t2.C, S=pw.require(t1.B + t2.D,t2.id)),
 include_id=False)

A  | t2_C | S
11 | 11   | 322
12 | 12   | 324
13 |      |
13 |      |

join_outer(other, *on, id=None)

Outer-joins two tables or join results.

• Parameters
  • other (Joinable) – Table or join result.
  • *on – Columns to join, syntax self.col1 == other.col2
  • id (OptionalColumnReference) – optional id column of the result

Remarks: args cannot contain id column from either of tables, as the result table has id column with auto-generated ids; it can be selected by assigning it to a column with defined name (passed in kwargs)

Behavior:

• for rows from the left side that were not matched with the right side, missing values on the right are replaced with None
• for rows from the right side that were not matched with the left side, missing values on the left are replaced with None
• for rows that were matched the behavior is the same as that of an inner join.

• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
         | A  | B
       1 | 11 | 111
       2 | 12 | 112
       3 | 13 | 113
       4 | 13 | 114
     '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
         | C  | D
       1 | 11 | 211
       2 | 12 | 212
       3 | 14 | 213
       4 | 14 | 214
     '''
 )
 pw.debug.compute_and_print(t1.join_outer(t2, t1.A == t2.C
 ).select(t1.A, t2_C=t2.C, S=pw.require(t1.B + t2.D,t1.id,t2.id)),
 include_id=False)

A  | t2_C | S
   | 14   |
   | 14   |
11 | 11   | 322
12 | 12   | 324
13 |      |
13 |      |

join_right(other, *on, id=None)

Outer-joins two tables or join results.

• Parameters
  • other (Joinable) – Table or join result.
  • *on – Columns to join, syntax self.col1 == other.col2
  • id (OptionalColumnReference) – optional id column of the result

Remarks: args cannot contain id column from either of tables, as the result table has id column with auto-generated ids; it can be selected by assigning it to a column with defined name (passed in kwargs)

Behavior:

• rows from the left side that were not matched with the right side are skipped
• for rows from the right side that were not matched with the left side, missing values on the left are replaced with None
• for rows that were matched the behavior is the same as that of an inner join.

• Returns
  an object on which .select() may be called to extract relevant columns from the result of the join.
• Return type
  JoinResult

Example:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
         | A  | B
       1 | 11 | 111
       2 | 12 | 112
       3 | 13 | 113
       4 | 13 | 114
     '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
         | C  | D
       1 | 11 | 211
       2 | 12 | 212
       3 | 14 | 213
       4 | 14 | 214
     '''
 )
 pw.debug.compute_and_print(t1.join_right(t2, t1.A == t2.C
 ).select(t1.A, t2_C=t2.C, S=pw.require(pw.coalesce(t1.B,0) + t2.D,t1.id)),
 include_id=False)

A  | t2_C | S
   | 14   |
   | 14   |
11 | 11   | 322
12 | 12   | 324

• Returns
  OuterJoinResult object

promise_universe_is_equal_to(other)

Asserts to Pathway that an universe of self is a subset of universe of each of the others.

Semantics: Used in situations where Pathway cannot deduce one universe being a subset of another.

Returns: None

Note: The assertion works in place.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | pet
 1 | Dog
 7 | Cat
 ''')
 t2 = pw.debug.parse_to_table('''
   | age
 1 | 10
 7 | 3
 ''')
 t1 = t1.promise_universe_is_equal_to(t2)
 t3 = t1 + t2
 pw.debug.compute_and_print(t3, include_id=False)

pet | age
Cat | 3
Dog | 10

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

promise_universe_is_subset_of(other)

Asserts to Pathway that an universe of self is a subset of universe of each of the other.

Semantics: Used in situations where Pathway cannot deduce one universe being a subset of another.

Returns: self

Note: The assertion works in place.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 1
 2 | 9   | Bob   | 1
 3 | 8   | Alice | 2
 ''')
 t2 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 30
 ''').promise_universe_is_subset_of(t1)
 t3 = t1 << t2
 pw.debug.compute_and_print(t3, include_id=False)

age | owner | pet
8   | Alice | 2
9   | Bob   | 1
10  | Alice | 30

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

promise_universes_are_disjoint(other)

Asserts to Pathway that an universe of self is disjoint from universe of other.

Semantics: Used in situations where Pathway cannot deduce universes are disjoint.

Returns: self

Note: The assertion works in place.

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 1
 2 | 9   | Bob   | 1
 3 | 8   | Alice | 2
 ''')
 t2 = pw.debug.parse_to_table('''
    | age | owner | pet
 11 | 11  | Alice | 30
 12 | 12  | Tom   | 40
 ''').promise_universes_are_disjoint(t1)
 t3 = t1.concat(t2)
 pw.debug.compute_and_print(t3, include_id=False)

age | owner | pet
8   | Alice | 2
9   | Bob   | 1
10  | Alice | 1
11  | Alice | 30
12  | Tom   | 40

• Return type
  TypeVar(SelfTableLike, bound= TableLike)

class pathway.Pointer()

Pointer to row type.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age | owner | pet
 10  | Alice | dog
 9   | Bob   | dog
 8   | Alice | cat
 7   | Bob   | dog
 ''')
 t2 = t1.select(col=t1.id)
 t2.schema['col'] is pw.Pointer

True

class pathway.Schema()

Base class to inherit from when creating schemas. All schemas should be subclasses of this one.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
    age  owner  pet
 1   10  Alice  dog
 2    9    Bob  dog
 3    8  Alice  cat
 4    7    Bob  dog''')
 t1.schema.as_dict()

{'age': <class 'int'>, 'owner': <class 'str'>, 'pet': <class 'str'>}

 issubclass(t1.schema, pw.Schema)

True

 class NewSchema(pw.Schema):
   foo: int
 SchemaSum = NewSchema | t1.schema
 SchemaSum.as_dict()

{'age': <class 'int'>, 'owner': <class 'str'>, 'pet': <class 'str'>, 'foo': <class 'int'>}

class pathway.StatsMonitoringLevel()

Specifies the verbosity of computation graph monitoring mechanism

NONE: No monitoring

IN_OUT: Monitor input and output latency. The latency is measured as the difference between the time when the operator processed the data and the time when pathway acquired the data.

ALL: Monitor latency for each operator in the execution graph latency. The latency is measured as the difference between the time when the operator processed the data and the time when pathway acquired the data.

class pathway.Table(columns, universe, pk_columns={}, schema=None, id_column=None)

Collection of named columns over identical universes.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age | owner | pet
 10  | Alice | dog
 9   | Bob   | dog
 8   | Alice | cat
 7   | Bob   | dog
 ''')
 isinstance(t1, pw.Table)

True

asof_join(other, t_left, t_right, *on, mode, defaults={})

Perform an ASOF join of two tables

• Parameters
  • other (Table) – Table to join with self, both must contain a column val
  • t_left (ColumnExpression) – time-like column expression to do the join against
  • t_right (ColumnExpression) – time-like column expression to do the join against
  • on (ColumnBinaryOpExpression) – a list of column expressions. Each must have == as the top level operation and be of the form LHS: ColumnReference == RHS: ColumnReference.
  • mode (JoinMode) – mode of the join (LEFT, RIGHT, FULL)
  • defaults (DictColumnReference, Any) – dictionary column-> default value. Entries in the resulting table that not have a predecessor in the join will be set to this default value. If no default is provided, None will be used.

Example:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
         | K | val |  t
     1   | 0 | 1   |  1
     2   | 0 | 2   |  4
     3   | 0 | 3   |  5
     4   | 0 | 4   |  6
     5   | 0 | 5   |  7
     6   | 0 | 6   |  11
     7   | 0 | 7   |  12
     8   | 1 | 8   |  5
     9   | 1 | 9   |  7
 '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
             | K | val | t
     21   | 1 | 7  | 2
     22   | 1 | 3  | 8
     23   | 0 | 0  | 2
     24   | 0 | 6  | 3
     25   | 0 | 2  | 7
     26   | 0 | 3  | 8
     27   | 0 | 9  | 9
     28   | 0 | 7  | 13
     29   | 0 | 4  | 14
     '''
 )
 res = t1.asof_join(
     t2,
     t1.t,
     t2.t,
     t1.K == t2.K,
     mode=pw.JoinMode.LEFT,
     defaults={t2.val: -1},
 ).select(
     pw.this.shard_key,
     pw.this.t,
     val_left=t1.val,
     val_right=t2.val,
     sum=t1.val + t2.val,
 )
 pw.debug.compute_and_print(res, include_id=False)

shard_key | t  | val_left | val_right | sum
0         | 1  | 1        | -1        | 0
0         | 4  | 2        | 6         | 8
0         | 5  | 3        | 6         | 9
0         | 6  | 4        | 6         | 10
0         | 7  | 5        | 6         | 11
0         | 11 | 6        | 9         | 15
0         | 12 | 7        | 9         | 16
1         | 5  | 8        | 7         | 15
1         | 7  | 9        | 7         | 16

concat(*others)

Concats self with every other ∊ others.

Semantics:

• result.columns == self.columns == other.columns
• result.id == self.id ∪ other.id

if self.id and other.id collide, throws an exception.

Requires:

• other.columns == self.columns
• self.id disjoint with other.id

• Parameters
  other – the other table.
• Returns
  The concatenated table. Id’s of rows from original tables are preserved.
• Return type
  Table

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | age | owner | pet
 1 | 10  | Alice | 1
 2 | 9   | Bob   | 1
 3 | 8   | Alice | 2
 ''')
 t2 = pw.debug.parse_to_table('''
    | age | owner | pet
 11 | 11  | Alice | 30
 12 | 12  | Tom   | 40
 ''')
 pw.universes.promise_are_pairwise_disjoint(t1, t2)
 t3 = t1.concat(t2)
 pw.debug.compute_and_print(t3, include_id=False)

age | owner | pet
8   | Alice | 2
9   | Bob   | 1
10  | Alice | 1
11  | Alice | 30
12  | Tom   | 40

concat_reindex(*tables)

Concatenate contents of several tables.

This is similar to PySpark union. All tables must have the same schema. Each row is reindexed.

• Parameters
  tables (Table) – List of tables to concatenate. All tables must have the same schema.
• Returns
  The concatenated table. It will have new, synthetic ids.
• Return type
  Table

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | pet
 1 | Dog
 7 | Cat
 ''')
 t2 = pw.debug.parse_to_table('''
   | pet
 1 | Manul
 8 | Octopus
 ''')
 t3 = t1.concat_reindex(t2)
 pw.debug.compute_and_print(t3, include_id=False)

pet
Cat
Dog
Manul
Octopus

copy()

Returns a copy of a table.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age | owner | pet
 10  | Alice | dog
 9   | Bob   | dog
 8   | Alice | cat
 7   | Bob   | dog
 ''')
 t2 = t1.copy()
 pw.debug.compute_and_print(t2, include_id=False)

age | owner | pet
7   | Bob   | dog
8   | Alice | cat
9   | Bob   | dog
10  | Alice | dog

 t1 is t2

False

• Return type
  Table

difference(other)

Restrict self universe to keys not appearing in the other table.

• Parameters
  other (Table) – table with ids to remove from self.
• Returns
  table with restricted universe, with the same set of columns
• Return type
  Table

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | age  | owner  | pet
 1 | 10   | Alice  | 1
 2 | 9    | Bob    | 1
 3 | 8    | Alice  | 2
 ''')
 t2 = pw.debug.parse_to_table('''
   | cost
 2 | 100
 3 | 200
 4 | 300
 ''')
 t3 = t1.difference(t2)
 pw.debug.compute_and_print(t3, include_id=False)

age | owner | pet
10  | Alice | 1

dtypes()

Return the types of the columns as a dictionary.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age | owner | pet
 10  | Alice | dog
 9   | Bob   | dog
 8   | Alice | cat
 7   | Bob   | dog
 ''')
 t1.dtypes()

{'age': <class 'int'>, 'owner': <class 'str'>, 'pet': <class 'str'>}

 t1.schema['age']

<class 'int'>

static empty(**kwargs)

Creates an empty table with a schema specified by kwargs.

• Parameters
  kwargs (NewType(DType, type)) – Dict whose keys are column names and values are column types.
• Returns
  Created empty table.
• Return type
  Table

Example:

 import pathway as pw
 t1 = pw.Table.empty(age=float, pet=float)
 pw.debug.compute_and_print(t1, include_id=False)

age | pet

filter(filter_expression)

Filter a table according to filter condition.

• Parameters
  filter – ColumnExpression that specifies the filtering condition.
• Returns
  Result has the same schema as self and its ids are subset of self.id.
• Return type
  Table

Example:

 import pathway as pw
 vertices = pw.debug.parse_to_table('''
 label outdegree
     1         3
     7         0
 ''')
 filtered = vertices.filter(vertices.outdegree == 0)
 pw.debug.compute_and_print(filtered, include_id=False)

label | outdegree
7     | 0

flatten(*args, **kwargs)

Performs a flatmap operation on a column or expression given as a first argument. Datatype of this column or expression has to be iterable. Other columns specified in the method arguments are duplicated as many times as the length of the iterable.

It is possible to get ids of source rows by using table.id column, e.g. table.flatten(table.column_to_be_flattened, original_id = table.id).

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | pet  |  age
 1 | Dog  |   2
 7 | Cat  |   5
 ''')
 t2 = t1.flatten(t1.pet)
 pw.debug.compute_and_print(t2, include_id=False)

pet
C
D
a
g
o
t

 t3 = t1.flatten(t1.pet, t1.age)
 pw.debug.compute_and_print(t3, include_id=False)

pet | age
C   | 5
D   | 2
a   | 5
g   | 2
o   | 2
t   | 5

• Return type
  Table

static from_columns(*args, **kwargs)

Build a table from columns.

All columns must have the same ids. Columns’ names must be pairwise distinct.

• Parameters
  • args (ColumnReference) – List of columns.
  • kwargs (ColumnReference) – Columns with their new names.
• Returns
  Created table.
• Return type
  Table

Example:

 import pathway as pw
 t1 = pw.Table.empty(age=float, pet=float)
 t2 = pw.Table.empty(foo=float, bar=float)
 t3 = pw.Table.from_columns(t1.pet, qux=t2.foo)
 pw.debug.compute_and_print(t3, include_id=False)

pet | qux

groupby(*args, id=None)

Groups table by columns from args.

NOTE: Usually followed by .reduce() that aggregates the result and returns a table.

• Parameters
  • args (ColumnReference) – columns to group by.
  • id (OptionalColumnReference) – if provided, is the column used to set id’s of the rows of the result
• Returns
  Groupby object.
• Return type
  GroupedTable

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age | owner | pet
 10  | Alice | dog
 9   | Bob   | dog
 8   | Alice | cat
 7   | Bob   | dog
 ''')
 t2 = t1.groupby(t1.pet, t1.owner).reduce(t1.owner, t1.pet, ageagg=pw.reducers.sum(t1.age))
 pw.debug.compute_and_print(t2, include_id=False)

owner | pet | ageagg
Alice | cat | 8
Alice | dog | 10
Bob   | dog | 16

having(*indexers)

Removes rows so that indexed.ix(indexer) is possible when some rows are missing, for each indexer in indexers

• Return type
  Table

property id(: ColumnReferenc )

Get reference to pseudocolumn containing id’s of a table.

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
 age | owner | pet
 10  | Alice | dog
 9   | Bob   | dog
 8   | Alice | cat
 7   | Bob   | dog
 ''')
 t2 = t1.select(ids = t1.id)
 pw.debug.compute_and_print(t2.select(test=t2.id == t2.ids), include_id=False)

test
True
True
True
True

• Return type
  ColumnReference

interpolate(timestamp, *values, mode=InterpolateMode.LINEAR)

Interpolates missing values in a column using the previous and next values based on a timestamps column.

• Parameters
  • timestamp (ColumnReference) – Reference to the column containing timestamps.
  • *values (ColumnReference) – References to the columns containing values to be interpolated.
  • mode (InterpolateMode*, *optional) – The interpolation mode. Currently, only InterpolateMode.LINEAR is supported. Default is InterpolateMode.LINEAR.
• Returns
  A new table with the interpolated values.
• Return type
  Table
• Raises
  ValueError – If the columns are not ColumnReference or if the interpolation mode is not supported.

NOTE:

• The interpolation is performed based on linear interpolation between the previous and next values.
• If a value is missing at the beginning or end of the column, no interpolation is performed.

Example:

 import pathway as pw
 table = pw.debug.table_from_markdown('''
 timestamp | values_a | values_b
 1         | 1        | 10
 2         |          |
 3         | 3        |
 4         |          |
 5         |          |
 6         | 6        | 60
 ''')
 table = table.interpolate(pw.this.timestamp, pw.this.values_a, pw.this.values_b)
 pw.debug.compute_and_print(table, include_id=False)

timestamp | values_a | values_b
1         | 1        | 10
2         | 2.0      | 20.0
3         | 3        | 30.0
4         | 4.0      | 40.0
5         | 5.0      | 50.0
6         | 6        | 60

intersect(*tables)

Restrict self universe to keys appearing in all of the tables.

• Parameters
  tables (Table) – tables keys of which are used to restrict universe.
• Returns
  table with restricted universe, with the same set of columns
• Return type
  Table

Example:

 import pathway as pw
 t1 = pw.debug.parse_to_table('''
   | age  | owner  | pet
 1 | 10   | Alice  | 1
 2 | 9    | Bob    | 1
 3 | 8    | Alice  | 2
 ''')
 t2 = pw.debug.parse_to_table('''
   | cost
 2 | 100
 3 | 200
 4 | 300
 ''')
 t3 = t1.intersect(t2)
 pw.debug.compute_and_print(t3, include_id=False)

age | owner | pet
8   | Alice | 2
9   | Bob   | 1

interval_join(other, self_time_expression, other_time_expression, lower_bound, upper_bound, *on)

Performs an interval join of self with other using a time difference and join expressions. If self_time_expression + lower_bound <= other_time_expression <= self_time_expression + upper_bound and conditions in on are satisfied, the rows are joined.

• Parameters
  • other (Table) – the right side of a join.
  • self_time_expression (ColumnExpression) – time expression in self.
  • other_time_expression (ColumnExpression) – time expression in other.
  • lower_bound (int | float | timedelta) – a lower bound on time difference between other_time_expression and self_time_expression.
  • upper_bound (int | float | timedelta) – an upper bound on time difference between other_time_expression and self_time_expression.
  • on (ColumnBinaryOpExpression) – a list of column expressions. Each must have == as the top level operation and be of the form LHS: ColumnReference == RHS: ColumnReference.
• Returns
  a result of the interval join. A method .select() can be called on it to extract relevant columns from the result of a join.
• Return type
  IntervalJoinResult

Examples:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
     | t
   1 | 3
   2 | 4
   3 | 5
   4 | 11
 '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
     | t
   1 | 0
   2 | 1
   3 | 4
   4 | 7
 '''
 )
 t3 = t1.interval_join(t2, t1.t, t2.t, -2, 1).select(
     left_t=t1.t, right_t=t2.t
 )
 pw.debug.compute_and_print(t3, include_id=False)

left_t | right_t
3      | 1
3      | 4
4      | 4
5      | 4

 t1 = pw.debug.table_from_markdown(
     '''
     | a | t
   1 | 1 | 3
   2 | 1 | 4
   3 | 1 | 5
   4 | 1 | 11
   5 | 2 | 2
   6 | 2 | 3
   7 | 3 | 4
 '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
     | b | t
   1 | 1 | 0
   2 | 1 | 1
   3 | 1 | 4
   4 | 1 | 7
   5 | 2 | 0
   6 | 2 | 2
   7 | 4 | 2
 '''
 )
 t3 = t1.interval_join(t2, t1.t, t2.t, -2, 1, t1.a == t2.b).select(
     t1.a, left_t=t1.t, right_t=t2.t
 )
 pw.debug.compute_and_print(t3, include_id=False)

a | left_t | right_t
1 | 3      | 1
1 | 3      | 4
1 | 4      | 4
1 | 5      | 4
2 | 2      | 0
2 | 2      | 2
2 | 3      | 2

interval_join_left(other, self_time_expression, other_time_expression, lower_bound, upper_bound, *on)

Performs an interval left join of self with other using a time difference and join expressions. If self_time_expression + lower_bound <= other_time_expression <= self_time_expression + upper_bound and conditions in on are satisfied, the rows are joined. Rows from the left side that haven’t been matched with the right side are returned with missing values on the right side replaced with None.

• Parameters
  • other (Table) – the right side of the join.
  • self_time_expression (ColumnExpression) – time expression in self.
  • other_time_expression (ColumnExpression) – time expression in other.
  • lower_bound (int | float | timedelta) – a lower bound on time difference between other_time_expression and self_time_expression.
  • upper_bound (int | float | timedelta) – an upper bound on time difference between other_time_expression and self_time_expression.
  • on (ColumnBinaryOpExpression) – a list of column expressions. Each must have == as the top level operation and be of the form LHS: ColumnReference == RHS: ColumnReference.
• Returns
  a result of the interval join. A method .select() can be called on it to extract relevant columns from the result of a join.
• Return type
  IntervalJoinResult

Examples:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
     | t
   1 | 3
   2 | 4
   3 | 5
   4 | 11
 '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
     | t
   1 | 0
   2 | 1
   3 | 4
   4 | 7
 '''
 )
 t3 = t1.interval_join_left(t2, t1.t, t2.t, -2, 1).select(left_t=t1.t, right_t=t2.t)
 pw.debug.compute_and_print(t3, include_id=False)

left_t | right_t
3      | 1
3      | 4
4      | 4
5      | 4
11     |

 t1 = pw.debug.table_from_markdown(
     '''
     | a | t
   1 | 1 | 3
   2 | 1 | 4
   3 | 1 | 5
   4 | 1 | 11
   5 | 2 | 2
   6 | 2 | 3
   7 | 3 | 4
 '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
     | b | t
   1 | 1 | 0
   2 | 1 | 1
   3 | 1 | 4
   4 | 1 | 7
   5 | 2 | 0
   6 | 2 | 2
   7 | 4 | 2
 '''
 )
 t3 = t1.interval_join_left(t2, t1.t, t2.t, -2, 1, t1.a == t2.b).select(
     t1.a, left_t=t1.t, right_t=t2.t
 )
 pw.debug.compute_and_print(t3, include_id=False)

a | left_t | right_t
1 | 3      | 1
1 | 3      | 4
1 | 4      | 4
1 | 5      | 4
1 | 11     |
2 | 2      | 0
2 | 2      | 2
2 | 3      | 2
3 | 4      |

interval_join_outer(other, self_time_expression, other_time_expression, lower_bound, upper_bound, *on)

Performs an interval outer join of self with other using a time difference and join expressions. If self_time_expression + lower_bound <= other_time_expression <= self_time_expression + upper_bound and conditions in on are satisfied, the rows are joined. Rows that haven’t been matched with the other side are returned with missing values on the other side replaced with None.

• Parameters
  • other (Table) – the right side of the join.
  • self_time_expression (ColumnExpression) – time expression in self.
  • other_time_expression (ColumnExpression) – time expression in other.
  • lower_bound (int | float | timedelta) – a lower bound on time difference between other_time_expression and self_time_expression.
  • upper_bound (int | float | timedelta) – an upper bound on time difference between other_time_expression and self_time_expression.
  • on (ColumnBinaryOpExpression) – a list of column expressions. Each must have == as the top level operation and be of the form LHS: ColumnReference == RHS: ColumnReference.
• Returns
  a result of the interval join. A method .select() can be called on it to extract relevant columns from the result of a join.
• Return type
  IntervalJoinResult

Examples:

 import pathway as pw
 t1 = pw.debug.table_from_markdown(
     '''
     | t
   1 | 3
   2 | 4
   3 | 5
   4 | 11
 '''
 )
 t2 = pw.debug.table_from_markdown(
     '''
     | t
   1 | 0
   2 | 1
   3 | 4
   4 | 7
 '''
 )
 t3 = t1.interval_join_outer(t2, t1.t, t2.t, -2, 1).select(left_t=t1.t, right_t=t2.t)
 pw.debug.compute_and_print(t3, include_id=False)

left_t | right_t
       | 0
       | 7
3      | 1
3      | 4
4      | 4
5      | 4
11     |