Pathway API
Reference for all the Pathway classes and functions.
See Table API for the main Table class.
class pw.AsyncTransformer(input_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)
close()
Called once at the end. Proper place for cleanup.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
.
open()
Called before actual work. Suitable for one time setup.with_options(capacity=None, retry_strategy=None, cache_strategy=None)
Sets async options.- Parameters
- capacity (
Optional
[int
]) – maximum number of concurrent operations. - retry_strategy (
Optional
[AsyncRetryStrategy
]) – defines how failures will be handled.
- capacity (
- Returns
self
property result(: Table )
Resulting table.class pw.ClassArg(ref: RowReference, ptr: Pointer)
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)
pointer_from(*args, optional=False)
Pseudo-random hash of its argument. Produces pointer types. Applied value-wise.class pw.ColumnExpression()
get(index, default=None)
Extracts element at index from an object. The object has to be a Tuple. If no element is present at index, it returns value specified by a default parameter.- Parameters
- index (
Union
[ColumnExpression
,int
]) – Position to extract element at. - default (
Union
[ColumnExpression
,None
,int
,float
,str
,bytes
,bool
,BasePointer
,datetime
,timedelta
,ndarray
,Tuple
[Union
[None
,int
,float
,str
,bytes
,bool
,BasePointer
,datetime
,timedelta
,ndarray
,Tuple
[Value,...
]],...
]]) – Value returned when no element is at position index.
- index (
Example:
import pathway as pw
t1 = pw.debug.table_from_markdown(
'''
| a | b | c
1 | 3 | 2 | 2
2 | 4 | 1 | 0
3 | 7 | 3 | 1
'''
)
t2 = t1.with_columns(tup=pw.make_tuple(pw.this.a, pw.this.b))
t3 = t2.select(
x=pw.this.tup.get(1),
y=pw.this.tup.get(3),
z=pw.this.tup.get(pw.this.c),
t=pw.this.tup.get(pw.this.c, default=100),
)
pw.debug.compute_and_print(t3, include_id=False)
to_string()
Changes the values to strings.Example:
import pathway as pw
t1 = pw.debug.parse_to_table('''
val
1
2
3
4''')
t1.schema.as_dict()
pw.debug.compute_and_print(t1, include_id=False)
t2 = t1.select(val = pw.this.val.to_string())
t2.schema.as_dict()
pw.debug.compute_and_print(t2.select(val=pw.this.val + "a"), include_id=False)
class pw.ColumnReference(*, column, table, name)
Inherits from ColumnExpression.
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)
isinstance(t1["owner"], pw.ColumnReference)
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
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
class pw.MonitoringLevel(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)
ALL( = 4 )
Monitor input connectors and latency for each operator in the execution graph. The latency is measured as the difference between the time when the operator processed the data and the time when pathway acquired the data.AUTO( = 0 )
Automatically sets IN_OUT in an interactive terminal and jupyter notebook. Sets NONE otherwise.AUTO_ALL( = 1 )
Automatically sets ALL in an interactive terminal and jupyter notebook. Sets NONE otherwise.IN_OUT( = 3 )
Monitor input connectors and 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.NONE( = 2 )
No monitoring.class pw.Pointer()
class pw.Schema()
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()
issubclass(t1.schema, pw.Schema)
class NewSchema(pw.Schema):
foo: int
SchemaSum = NewSchema | t1.schema
SchemaSum.as_dict()
class pw.SchemaProperties(append_only=False)
class pw.TableSlice(mapping, 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
''')
t1.slice.without("age").with_suffix("_col")
class pw.iterate_universe(table)
class pw.left(*args, **kwargs)
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, pw.left.pet == pw.right.pet, pw.left.owner == pw.right.owner).select(
age=pw.left.age, owner_name=pw.right.owner, size=pw.this.size
)
pw.debug.compute_and_print(t3, include_id=False)
class pw.right(*args, **kwargs)
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, pw.left.pet == pw.right.pet, pw.left.owner == pw.right.owner).select(
age=pw.left.age, owner_name=pw.right.owner, size=pw.this.size
)
pw.debug.compute_and_print(t3, include_id=False)
class pw.this(*args, **kwargs)
Example:
import pathway as pw
t1 = pw.debug.parse_to_table('''
age | owner | pet
10 | Alice | 1
9 | Bob | 1
8 | Alice | 2
''')
t2 = t1.select(pw.this.owner, pw.this.age)
pw.debug.compute_and_print(t2, include_id=False)
Functions
pw.apply(fun, *args, **kwargs)
Applies function to column expressions, column-wise. Output column type deduced from type-annotations of a function.Example:
import pathway as pw
def concat(left: str, right: str) -> str:
return left+right
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 = pw.apply(concat, t1.owner, t1.pet))
pw.debug.compute_and_print(t2, include_id=False)
pw.apply_async(fun, *args, **kwargs)
Applies function asynchronously to column expressions, column-wise. Output column type deduced from type-annotations of a function. Either a regular or async function can be passed.Example:
import pathway as pw
import asyncio
async def concat(left: str, right: str) -> str:
await asyncio.sleep(0.1)
return left+right
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 = pw.apply_async(concat, t1.owner, t1.pet))
pw.debug.compute_and_print(t2, include_id=False)
pw.apply_with_type(fun, ret_type, *args, **kwargs)
Applies function to column expressions, column-wise. Output column type is provided explicitly.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''')
t2 = t1.select(col = pw.apply_with_type(lambda left, right: left+right, str, t1.owner, t1.pet))
pw.debug.compute_and_print(t2, include_id=False)
pw.assert_table_has_schema(table, schema, *, allow_superset=False, ignore_primary_keys=True)
Asserts that the schema of the table is equivalent to the schema given as an argument.- Parameters
- schema. (table - Table for which we are asserting) –
- Schema (schema -) –
- has. (which we assert that the Table) –
- True (ignore_primary_keys - if) – in schema.
- columns (the columns* of *the table can be a superset of) – in schema.
- True – have the same primary keys. The default value is True.
- schema (the assert won't check whether table and) – have the same primary keys. The default value is True.
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(pw.this.owner, age = pw.cast(float, pw.this.age))
schema = pw.schema_builder(
{"age": pw.column_definition(dtype=float), "owner": pw.column_definition(dtype=str)}
)
pw.assert_table_has_schema(t2, schema)
pw.attribute(func, **kwargs)
Decorator for creation of attributes.Example:
import pathway as pw
@pw.transformer
class simple_transformer:
class table(pw.ClassArg):
arg = pw.input_attribute()
@pw.attribute
def attr(self) -> float:
return self.arg*2
@pw.output_attribute
def ret(self) -> float:
return self.attr + 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)
pw.cast(target_type, col)
Changes the type of the column to target_type and converts the data of this columnExample:
import pathway as pw
t1 = pw.debug.parse_to_table('''
val
1 10
2 9
3 8
4 7''')
t1.schema.as_dict()
pw.debug.compute_and_print(t1, include_id=False)
t2 = t1.select(val = pw.cast(float, t1.val))
t2.schema.as_dict()
pw.debug.compute_and_print(t2, include_id=False)
pw.coalesce(*args)
For arguments list arg_1, arg_2, …, arg_n returns first not-None value.Example:
import pathway as pw
t1 = pw.debug.parse_to_table('''
colA colB
| 10
2 |
|
4 | 7''')
t2 = t1.select(t1.colA, t1.colB, col=pw.coalesce(t1.colA, t1.colB))
pw.debug.compute_and_print(t2, include_id=False)
pw.column_definition(*, primary_key=False, default_value=undefined, dtype=None, name=None)
Creates column definition- Parameters
- primary_key (
bool
) – should column be a part of a primary key. - default_value (
Optional
[Any
]) – default valuee replacing blank entries. The default value of the column must be specified explicitly, otherwise there will be no default value. - dtype (
Optional
[Any
]) – data type. When used in schema class, will be deduced from the type annotation. - name (
Optional
[str
]) – name of a column. When used in schema class, will be deduced from the attribute name.
- primary_key (
- Returns
Column definition.
Example:
import pathway as pw
class NewSchema(pw.Schema):
key: int = pw.column_definition(primary_key=True)
timestamp: str = pw.column_definition(name="@timestamp")
data: str
NewSchema.as_dict()
pw.declare_type(target_type, col)
Used to change the type of a column to a particular type. Disclaimer: it only changes type in a schema, it does not affect values stored.Example:
import pathway as pw
t1 = pw.debug.parse_to_table('''
val
1 10
2 9.5
3 8
4 7''')
t1.schema.as_dict()
t2 = t1.filter(t1.val == pw.cast(int, t1.val))
t2.schema.as_dict()
t3 = t2.select(val = pw.declare_type(int, t2.val))
t3.schema.as_dict()
pw.if_else(if_clause, then_clause, else_clause)
Equivalent to:if (if_clause):
return (then_clause)
else:
return (else_clause)
Example:
import pathway as pw
t1 = pw.debug.parse_to_table('''
colA colB
1 | 0
2 | 2
6 | 3''')
t2 = t1.select(res = pw.if_else(t1.colB != 0, t1.colA // t1.colB, 0))
pw.debug.compute_and_print(t2, include_id=False)
pw.input_attribute(type=<class 'float'>)
Returns new input_attribute. To be used inside 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) -> float:
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)
pw.input_method(type=<class 'float'>)
Decorator for defining input methods in class transformers.Example:
import pathway as pw
@pw.transformer
class first_transformer:
class table(pw.ClassArg):
a: float = pw.input_attribute()
@pw.method
def fun(self, arg) -> int:
return self.a * arg
@pw.transformer
class second_transformer:
class table(pw.ClassArg):
m = pw.input_method(int)
@pw.output_attribute
def val(self):
return self.m(2)
t1 = pw.debug.parse_to_table('''
age
10
9
8
7''')
t2 = first_transformer(table=t1.select(a=t1.age)).table
t2.schema.as_dict()
t3 = second_transformer(table=t2.select(m=t2.fun)).table
pw.debug.compute_and_print(t1 + t3, include_id=False)
pw.iterate(func, iteration_limit=None, **kwargs)
Iterate function until fixed point. Function has to take only named arguments, Tables, and return a dict of Tables. Initial arguments to function are passed through kwargs.Example:
import pathway as pw
def collatz_transformer(iterated):
def collatz_step(x: int) -> int:
if x == 1:
return 1
elif x % 2 == 0:
return x / 2
else:
return 3 * x + 1
new_iterated = iterated.select(val=pw.apply(collatz_step, iterated.val))
return dict(iterated=new_iterated)
tab = pw.debug.parse_to_table('''
val
1
2
3
4
5
6
7
8''')
ret = pw.iterate(collatz_transformer, iterated=tab).iterated
pw.debug.compute_and_print(ret, include_id=False)
pw.make_tuple(*args)
Creates a tuple from the provided expressions.- Parameters
args (Union
[ColumnExpression
,None
,int
,float
,str
,bytes
,bool
,BasePointer
,datetime
,timedelta
,ndarray
,Tuple
[Union
[None
,int
,float
,str
,bytes
,bool
,BasePointer
,datetime
,timedelta
,ndarray
,Tuple
[Value,...
]],...
]]) – a list of expressions to be put in a tuple - Returns
tuple
NOTE: * Each cell in the output column will be a tuple containing the corresponding values from the input columns.
- The order of values in each tuple will match the order of the input columns.
- If any of the input columns have missing values, the resulting tuples will contain None for those positions.
Example:
import pathway as pw
table = pw.debug.table_from_markdown(
'''
A | B | C
1 | 10 | a
2 | 20 |
3 | 30 | c
'''
)
table_with_tuple = table.select(res=pw.make_tuple(pw.this.A, pw.this.B, pw.this.C))
pw.debug.compute_and_print(table_with_tuple, include_id=False)
pw.method(func, **kwargs)
Decorator for creation methods in class transformers.Example:
import pathway as pw
@pw.transformer
class simple_transformer:
class table(pw.ClassArg):
a: float = pw.input_attribute()
@pw.output_attribute
def b(self) -> float:
return self.fun(self.a)
@method
def fun(self, arg) -> float:
return self.a * arg
t1 = pw.debug.parse_to_table('''
age
10
9
8
7''')
t2 = simple_transformer(table=t1.select(a=t1.age)).table
t2.schema.as_dict()
pw.debug.compute_and_print(t1 + t2.select(t2.b), include_id=False)
pw.debug.compute_and_print(t1 + t2.select(out = t2.fun(t2.b)), include_id=False)
pw.numba_apply(fun, numba_signature, *args, **kwargs)
Applies function to column expressions, column-wise. Function has to be numba compilable.Currently only a few signatures are supported:
- function has to be unary or binary
- arguments and return type has to be either int64 or float64
Example:
import pathway as pw
t1 = pw.debug.parse_to_table('''
val
1 1
2 3
3 5
4 7''')
t2 = t1.select(col = pw.numba_apply(lambda x: x*x-2*x+1, "int64(int64,)", t1.val))
pw.debug.compute_and_print(t2, include_id=False)
pw.output_attribute(func, **kwargs)
Decorator for creation of output_attributes.Example:
import pathway as pw
@pw.transformer
class simple_transformer:
class table(pw.ClassArg):
arg = pw.input_attribute()
@pw.output_attribute
def ret(self) -> float:
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)
pw.pandas_transformer(output_schema, output_universe=None)
Decorator that turns python function operating on pandas.DataFrame into pathway transformer.Input universes are converted into input DataFrame indexes. The resulting index is treated as the output universe, so it must maintain uniqueness and be of integer type.
- Parameters
- output_schema (
Type
[Schema
]) – Schema of a resulting table. - output_universe (
Union
[int
,str
,None
]) – Index or name of an argument whose universe will be used in resulting table. Defaults to None.
- output_schema (
- Returns
Transformer that can be applied on Pathway tables.
Example:
import pathway as pw
input = pw.debug.table_from_markdown(
'''
| foo | bar
0 | 10 | 100
1 | 20 | 200
2 | 30 | 300
'''
)
class Output(pw.Schema):
sum: int
@pw.pandas_transformer(output_schema=Output)
def sum_cols(t: pd.DataFrame) -> pd.DataFrame:
return pd.DataFrame(t.sum(axis=1))
output = sum_cols(input)
pw.debug.compute_and_print(output, include_id=False)
pw.require(val, *deps)
Returns val iff every dep in deps is not-None. Returns None otherwise.Example:
import pathway as pw
t1 = pw.debug.parse_to_table('''
colA colB
| 10
2 |
|
4 | 7''')
t2 = t1.select(t1.colA, t1.colB, col=pw.require(t1.colA + t1.colB, t1.colA, t1.colB))
pw.debug.compute_and_print(t2, include_id=False)
pw.run(debug=False, monitoring_level=MonitoringLevel.AUTO, with_http_server=False, default_logging=True, persistence_config=None)
Runs the computation graph.- Parameters
- debug (
bool
) – enable output out of table.debug() operators - monitoring_level (
MonitoringLevel
) – the verbosity of stats monitoring mechanism. One of pathway.MonitoringLevel.NONE, pathway.MonitoringLevel.IN_OUT, pathway.MonitoringLevel.ALL. If unset, pathway will choose between NONE and IN_OUT based on output interactivity. - with_http_server (
bool
) – whether to start a http server with runtime metrics. Learn more in a tutorial . - default_logging (
bool
) – whether to allow pathway to set its own logging handler. Set it to False if you want to set your own logging handler.
- debug (
pw.schema_builder(columns, *, name=None, properties=SchemaProperties(append_only=False))
Allows to build schema inline, from a dictionary of column definitions.- Parameters
- columns (
Dict
[str
,ColumnDefinition
]) – dictionary of column definitions. - name (
Optional
[str
]) – schema name. - properties (
SchemaProperties
) – schema properties.
- columns (
- Returns
Schema
Example:
import pathway as pw
schema = pw.schema_builder(columns={
'key': pw.column_definition(dtype=int, primary_key=True),
'data': pw.column_definition(dtype=int, default_value=0)
}, name="my_schema")
schema.as_dict()
pw.schema_from_types(_name=None, **kwargs)
Constructs schema from kwargs: field=type.Example:
import pathway as pw
s = pw.schema_from_types(foo=int, bar=str)
s.as_dict()
issubclass(s, pw.Schema)
pw.sql(query, **kwargs)
Run a SQL query on Pathway tables.- Parameters
- query (
str
) – the SQL query to execute. - kwargs (
Table
) – the association name: table used for the execution of the SQL query. Each name:table pair links a Pathway table to a table name used in the SQL query.
- query (
Example:
import pathway as pw
t = pw.debug.table_from_markdown(
"""
A | B
1 | 2
4 | 3
4 | 7
"""
)
ret = pw.sql("SELECT * FROM tab WHERE A<B", tab=t)
pw.debug.compute_and_print(ret, include_id=False)
Supported SQL keywords and operations: SELECT, WHERE, boolean expressions, arithmetic operations, GROUP BY, HAVING, AS (alias), UNION, INTERSECTION, JOIN, and WITH.
Table and column names are case-sensitive.
Specificities of Pathway:
- id is a reserved key word for columns, every Pathway table has a special column id. This column is not captured by * expressions in SQL.
- Order of columns might not be preserved with respect to SELECT query.
- Pathway reducers (pw.count, pw.sum, etc.) aggregate over None values, while SQL aggregation functions (COUNT, SUM, etc.) skip NULL values.
- UNION requires matching column names.
- INTERSECT requires matching column names.
Limited support:
- Subqueries are supported but fragile – they depend on a set of query rewriting routines from the sqlglot library.
- Additionally, using the id column in subqueries is fragile.
- LIKE, ANY, ALL, EXISTS are not supported, or only supported in a very weak state.
Unsupported operations:
- ordering operations: ORDER BY, LIMIT, SELECT TOP
- INSERT INTO (Pathway tables are immutable)
- Pathway does not support anonymous columns: they might work but we do not guarantee their behavior.
- INTERSECT does not support INTERSECT ALL.
- COALESCE, IFNULL are not supported.
- FULL JOIN and NATURAL JOIN are not supported.
- CAST is not supported
pw.transformer(cls)
Decorator that wraps the outer class when defining 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) -> float:
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)
pw.udf(fun)
Create a Python UDF (universal data function) out of a callable.The output type of the UDF is determined based on its type annotation.
Example:
import pathway as pw
@pw.udf
def concat(left: str, right: str) -> str:
return left+right
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 = concat(t1.owner, t1.pet))
pw.debug.compute_and_print(t2, include_id=False)
pw.udf_async(fun=None, *, capacity=None, retry_strategy=None, cache_strategy=None)
Create a Python asynchronous UDF (universal data function) out of a callable.Output column type deduced from type-annotations of a function. Can be applied to a regular or asynchronous function.
Example:
import pathway as pw
import asyncio
@pw.udf_async
async def concat(left: str, right: str) -> str:
await asyncio.sleep(0.1)
return left+right
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 = concat(t1.owner, t1.pet))
pw.debug.compute_and_print(t2, include_id=False)
pw.unwrap(col)
Changes the type of the column from Optional\[T\] to T. If there is any None in the column this operation will raise an exception.Example:
import pathway as pw
t1 = pw.debug.parse_to_table('''
colA | colB
1 | 5
2 | 9
3 | None
4 | 15''')
t1.schema.as_dict()
pw.debug.compute_and_print(t1, include_id=False)
t2 = t1.filter(t1.colA < 3)
t2.schema.as_dict()
pw.debug.compute_and_print(t2, include_id=False)
t3 = t2.select(colB = pw.unwrap(t2.colB))
t3.schema.as_dict()
pw.debug.compute_and_print(t3, include_id=False)