Repository: ibis-project/ibis

Issue:
feat(table): add namespace kwarg to ibis.table
A user asked about how to specify a table namespace when creating an unbound table.

The `UnboundTable` op does accept a namespace argument, but we don't expose that option in `ibis/expr/api.py::table`.

We will run afoul of the dual-definition of `schema` here, but if the user has to first construct the namespace object, at least that will be slightly separated.  

Still not a good solution, really, to the schema/schema issue, but I don't know what is, short of breaking all ibis code.


Discussion:
cpcloud: Looks like rogue parenthesis decided it didn't need to exist!

Patch:
diff --git a/docs/_freeze/tutorials/open-source-software/apache-flink/1_single_feature/execute-results/html.json b/docs/_freeze/tutorials/open-source-software/apache-flink/1_single_feature/execute-results/html.json
index 8fbe5d676..5b7c560b1 100644
--- a/docs/_freeze/tutorials/open-source-software/apache-flink/1_single_feature/execute-results/html.json
+++ b/docs/_freeze/tutorials/open-source-software/apache-flink/1_single_feature/execute-results/html.json
@@ -1,8 +1,8 @@
 {
-  "hash": "77966c042cd21758f26e4ffe3d5697a6",
+  "hash": "624a12f4f47142f2c874c385de3e037f",
   "result": {
     "engine": "jupyter",
-    "markdown": "---\ntitle: 'A real-life use case: fraud detection'\n---\n\n\n\nImagine you’re a data scientist who works at a large bank. You have been tasked\nwith one of the most challenging problems in banking today: identifying\nfraudulent transactions. The bank receives transaction details from its credit\ncard customers in a Kafka topic, which include information about the transaction\ndate and time, transaction amount, transaction location, merchant, category of\npurchase, and so on. Given the nature of the data, you want to use Apache Flink\nfor its stream processing capabilities and to develop machine learning features\nthat can be used to identify fraud.\n\n\n### Prerequisites\n\n* Docker Compose: This tutorial uses Docker Compose to manage an Apache Kafka\nenvironment (including sample data generation) and a Flink cluster (for remote\nexecution). You can [download and install Docker Compose from the official\nwebsite](https://docs.docker.com/compose/install/).\n* JDK 11 release: Flink requires Java 11.\n* Python 3.9 or 3.10.\n* Follow [the setup tutorial](0_setup.qmd) to install the Flink backend for Ibis.\n* Clone the [example repository](https://github.com/ibis-project/ibis-flink-tutorial).\n\n\n### Spinning up the services using Docker Compose\n\nFrom your project directory, run `docker compose up -d` to create Kafka topics,\ngenerate sample data, and launch a Flink cluster in the background.\n\n\n\n::: {.callout-tip}\nRunning `docker compose up` with the `-d` flag runs it in\ndetached mode, where containers are run in the background. While this frees up\nthe terminal for you to run other commands, it also hides error messages.\n\nDepending on whether the container images are already available locally, setting\nup the containers may take anywhere from 10 seconds to a minute. If it's your\nfirst time running this command, it's best to run it in the foreground so that\nyou can monitor the progress of setup.\n:::\n\nThis should set up a `transaction` topic in the Kafka cluster that contains\nmessages that look like the following:\n\n```\n{ \"trans_date_trans_time\": \"2012-02-23 00:10:01\", \"cc_num\":\n4428780000000000000, \"merchant\": \"fraud_Olson, Becker and Koch\", \"category\":\n\"gas_transport\", \"amt\": 82.55, \"first\": \"Richard\", \"last\": \"Waters\", \"zipcode\":\n\"53186\", \"dob\": \"1/2/46\", \"trans_num\": \"dbf31d83eebdfe96d2fa213df2043586\",\n\"is_fraud\": 0, \"user_id\": 7109464218691269943 }\n```\n\n::: {.callout-warning}\nDo not proceed to the next section until messages are\nflowing into the `transaction` topic!\n:::\n\n\n### Connect to a Flink environment session\n\nWe can connect to a Flink environment session by creating a\n`pyflink.table.TableEnvironment` and passing this to Flink backend’s `connect`\nmethod. For this tutorial, we are going to use Flink in streaming mode.\n\n::: {#36ad947f .cell execution_count=2}\n``` {.python .cell-code}\nfrom pyflink.table import EnvironmentSettings, TableEnvironment\n\nimport ibis\n\nenv_settings = EnvironmentSettings.in_streaming_mode()\ntable_env = TableEnvironment.create(env_settings)\ntable_env.get_config().set(\"parallelism.default\", \"1\")\nconnection = ibis.flink.connect(table_env)\n```\n:::\n\n\nThe Kafka connector isn’t part of the binary distribution. In order to connect\nto a Kafka source/sink, we need to [download the JAR\nfile](https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/3.0.2-1.18/flink-sql-connector-kafka-3.0.2-1.18.jar)\nand manually add it into the classpath:\n\n\nNow that we’ve set up the Flink table environment, we’re ready to connect to\ndata!\n\n\n### Connect to a data source\n\nIn order to experiment with the data in the Kafka topic and create\ntransformations on top of it, we need to first define and connect to the data\nsource.\n\nWhile we’re dealing with a continuous stream of data here, Flink and Ibis\nabstract differences in the underlying implementation between tables and\nstreams, so that, conceptually, we can simply treat our Kafka topic as a table.\n\nTo connect to our `transaction` Kafka topic, we need to provide a table name,\nschema of the data, and [connector\nconfigurations](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/connectors/table/kafka/#connector-options).\nThe schema of the data must contain a subset of the fields in the actual Kafka\ntopic. Because this is a streaming job, we also want to define a watermark\nstrategy for the data source by specifying the timestamp column (`time_col`) and\na time duration during which late events are accepted (`allowed_delay`). (If you\nare not already familiar with these concepts, you can check out [Flink’s\ndocumentation](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/datastream/operators/windows/)\nfor more details.) Note that [Flink requires the timestamp column to be of data\ntype\nTIMESTAMP(3)](https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/create/#watermark).\n\n::: {#76f28ed6 .cell execution_count=4}\n``` {.python .cell-code}\nimport ibis\nimport ibis.expr.datatypes as dt\nimport ibis.expr.schema as sch\n\nsource_schema = sch.Schema(\n    {\n        \"user_id\": dt.int64,\n        \"trans_date_trans_time\": dt.timestamp(scale=3),\n        \"cc_num\": dt.int64,\n        \"amt\": dt.float64,\n        \"trans_num\": dt.str,\n        \"merchant\": dt.str,\n        \"category\": dt.str,\n        \"is_fraud\": dt.int32,\n        \"first\": dt.str,\n        \"last\": dt.str,\n        \"dob\": dt.str,\n        \"zipcode\": dt.str,\n    }\n)\n\n# Configure the source table with Kafka connector properties.\nsource_configs = {\n    \"connector\": \"kafka\",\n    \"topic\": \"transaction\",\n    \"properties.bootstrap.servers\": \"localhost:9092\",\n    \"properties.group.id\": \"consumer_group_0\",\n    \"scan.startup.mode\": \"earliest-offset\",\n    \"format\": \"json\",\n}\n\n# Create the source table using the defined schema, Kafka connector properties,\n# and set watermarking for real-time processing with a 15-second allowed\n# lateness.\nsource_table = connection.create_table(\n    \"transaction\",\n    schema=source_schema,\n    tbl_properties=source_configs,\n    watermark=ibis.watermark(\n        time_col=\"trans_date_trans_time\", allowed_delay=ibis.interval(seconds=15)\n    ),\n)\n```\n:::\n\n\nWe’re ready to write some transformations!\n\n\n### Create transformations\n\nWhich signs could be indicative of suspected fraud in a credit card? Oftentimes,\nwe’re looking for abnormalities in user behaviors, for example, an excessively\nlarge transaction amount, unusually frequent transactions during a short period\nof time, etc. Based on this, the average transaction amount and the total\ntransaction count over the past five hours may be useful features. Let’s write\nout each of these using Ibis API:\n\n::: {#ae15cda7 .cell execution_count=5}\n``` {.python .cell-code}\nuser_trans_amt_last_360m_agg = source_table[\n    source_table.user_id,\n    # Calculate the average transaction amount over the past six hours\n    source_table.amt.mean()\n    .over(\n        ibis.window(\n            group_by=source_table.user_id,\n            order_by=source_table.trans_date_trans_time,\n            range=(-ibis.interval(minutes=360), 0),\n        )\n    )\n    .name(\"user_mean_trans_amt_last_360min\"),\n    # Calculate the total transaction count over the past six hours\n    source_table.amt.count()\n    .over(\n        ibis.window(\n            group_by=source_table.user_id,\n            order_by=source_table.trans_date_trans_time,\n            range=(-ibis.interval(minutes=360), 0),\n        )\n    )\n    .name(\"user_trans_count_last_360min\"),\n    source_table.trans_date_trans_time,\n]\n```\n:::\n\n\n`over()` creates an [over\naggregation](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/sql/queries/over-agg/)\nin Flink, which computes an aggregated value for every input row. More\nspecifically, this means that an aggregation result is computed and emitted for\nevery new record flowing into the upstream Kafka topic.\n\nThe issue with over aggregation is that, if there is no new transaction for a\nspecific user during a time window, there would be no aggregation result written\nto the sink. In other words, the user would never show up in the result table if\nthey never made a transaction.\n\nAlternatively, we can compute aggregations using [Flink’s windowing table-valued\nfunctions](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/sql/queries/window-tvf/).\nThis allows more flexibility in defining windows and when results are computed\nand emitted into the sink. There are three types of windowing TVFs available in\nFlink: tumble, hop, and cumulate. Let’s define the same features with tumble\nwindows:\n\n::: {#61545165 .cell execution_count=6}\n``` {.python .cell-code}\nwindowed_stream = source_table.window_by(\n    time_col=source_table.trans_date_trans_time,\n).tumble(window_size=ibis.interval(minutes=360))\n\nuser_trans_amt_last_360m_agg_windowed_stream = windowed_stream.group_by(\n    [\"window_start\", \"window_end\", \"user_id\"]\n).agg(\n    user_mean_trans_amt_last_360min=windowed_stream.amt.mean(),\n    user_trans_count_last_360min=windowed_stream.amt.count(),\n)\n```\n:::\n\n\n### Connect to a data sink\n\nWe’re creating streaming jobs to continuously process upstream data, which could\nbe infinite. Therefore, we want to have the job continuously running and write\nresults into a data sink. Here, we’re simply going to write results into a\nseparate Kafka topic named `user_trans_amt_last_360min` for convenient\ndownstream processing.\n\nWe can define a data sink in virtually the same exact way in which we defined\nour data source:\n\n::: {#8a74bd5e .cell execution_count=7}\n``` {.python .cell-code}\nsink_schema = sch.Schema(\n    {\n        \"user_id\": dt.int64,\n        \"user_mean_trans_amt_last_360min\": dt.float64,\n        \"user_trans_count_last_360min\": dt.int64,\n        \"trans_date_trans_time\": dt.timestamp(scale=3),\n    }\n)\n\n# Configure the sink table with Kafka connector properties for writing results.\nsink_configs = {\n    \"connector\": \"kafka\",\n    \"topic\": \"user_trans_amt_last_360min\",\n    \"properties.bootstrap.servers\": \"localhost:9092\",\n    \"format\": \"json\",\n}\n\nsink_table = connection.create_table(\n    \"user_trans_amt_last_360min\",\n    schema=sink_schema,\n    tbl_properties=sink_configs,\n)\n```\n:::\n\n\nThe last step is to connect the pieces and actually write our query results into\nthe sink table that we had just created:\n\n::: {#dc3f0e4c .cell execution_count=8}\n``` {.python .cell-code}\nconnection.insert(\"user_trans_amt_last_360min\",\nuser_trans_amt_last_360m_agg)\n```\n\n::: {.cell-output .cell-output-display execution_count=8}\n```\n<pyflink.table.table_result.TableResult at 0x15455a5f0>\n```\n:::\n:::\n\n\nThis step is exactly the same for windowing TVFs:\n\n::: {#a92f69bb .cell execution_count=9}\n``` {.python .cell-code}\nsink_schema = sch.Schema(\n    {\n        \"window_start\": dt.timestamp(scale=3),\n        \"window_end\": dt.timestamp(scale=3),\n        \"user_id\": dt.int64,\n        \"user_mean_trans_amt_last_360min\": dt.float64,\n        \"user_trans_count_last_360min\": dt.int64,\n    }\n)\n\n# Configure the sink table with Kafka connector properties for writing results.\nsink_configs = {\n    \"connector\": \"kafka\",\n    \"topic\": \"user_trans_amt_last_360min_windowed\",\n    \"properties.bootstrap.servers\": \"localhost:9092\",\n    \"format\": \"json\",\n}\n\nsink_table = connection.create_table(\n    \"user_trans_amt_last_360min_windowed\",\n    schema=sink_schema,\n    tbl_properties=sink_configs,\n)\n\nconnection.insert(\n    \"user_trans_amt_last_360min_windowed\", user_trans_amt_last_360m_agg_windowed_stream\n)\n```\n\n::: {.cell-output .cell-output-display execution_count=9}\n```\n<pyflink.table.table_result.TableResult at 0x1545adc00>\n```\n:::\n:::\n\n\n### Expected output\n\nNow, if everything is working correctly, you should expect to see results being\nstreamed into the Kafka topic!\n\n::: {.callout-tip}\nYou can inspect the Kafka topic using the Python Kafka client\nif you have it installed or via console Kafka consumer:\n\n::: {#5608aed6 .cell execution_count=10}\n``` {.python .cell-code}\nfrom kafka import KafkaConsumer\n\nconsumer = KafkaConsumer(\n    \"user_trans_amt_last_360min\"\n)  # or \"user_trans_amt_last_360min_windowed\"\nfor _, msg in zip(range(10), consumer):\n    print(msg)\n```\n\n::: {.cell-output .cell-output-stdout}\n```\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=48125, timestamp=1706914332495, timestamp_type=0, key=None, value=b'{\"user_id\":-312211619698468596,\"user_mean_trans_amt_last_360min\":1507.78,\"user_trans_count_last_360min\":1,\"trans_date_trans_time\":\"2012-01-29 03:26:20\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=152, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=48126, timestamp=1706914332495, timestamp_type=0, key=None, value=b'{\"user_id\":-1298429917834413049,\"user_mean_trans_amt_last_360min\":182.82,\"user_trans_count_last_360min\":1,\"trans_date_trans_time\":\"2012-01-29 03:26:56\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=152, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=48127, timestamp=1706914332495, timestamp_type=0, key=None, value=b'{\"user_id\":-4266863151675529758,\"user_mean_trans_amt_last_360min\":2.829999999999986,\"user_trans_count_last_360min\":1,\"trans_date_trans_time\":\"2012-01-29 03:28:15\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=163, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=48128, timestamp=1706914332495, timestamp_type=0, key=None, value=b'{\"user_id\":8211595333002534503,\"user_mean_trans_amt_last_360min\":3.49,\"user_trans_count_last_360min\":1,\"trans_date_trans_time\":\"2012-01-29 03:28:42\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=149, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=48129, timestamp=1706914332495, timestamp_type=0, key=None, value=b'{\"user_id\":-8800824711477190666,\"user_mean_trans_amt_last_360min\":70.93999999999998,\"user_trans_count_last_360min\":2,\"trans_date_trans_time\":\"2012-01-29 03:28:51\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=163, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=48130, timestamp=1706914332495, timestamp_type=0, key=None, value=b'{\"user_id\":-6588488932184900316,\"user_mean_trans_amt_last_360min\":22.05999999999996,\"user_trans_count_last_360min\":1,\"trans_date_trans_time\":\"2012-01-29 03:29:12\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=163, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=48131, timestamp=1706914332495, timestamp_type=0, key=None, value=b'{\"user_id\":-494959696861882363,\"user_mean_trans_amt_last_360min\":53.07000000000003,\"user_trans_count_last_360min\":1,\"trans_date_trans_time\":\"2012-01-29 03:31:18\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=162, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=48132, timestamp=1706914332495, timestamp_type=0, key=None, value=b'{\"user_id\":6819872428711972947,\"user_mean_trans_amt_last_360min\":31.635,\"user_trans_count_last_360min\":2,\"trans_date_trans_time\":\"2012-01-29 03:32:20\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=151, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=48133, timestamp=1706914332495, timestamp_type=0, key=None, value=b'{\"user_id\":5518566883052153333,\"user_mean_trans_amt_last_360min\":5.799999999999996,\"user_trans_count_last_360min\":2,\"trans_date_trans_time\":\"2012-01-29 03:33:14\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=162, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=48134, timestamp=1706914332495, timestamp_type=0, key=None, value=b'{\"user_id\":2295915180219832597,\"user_mean_trans_amt_last_360min\":31.93,\"user_trans_count_last_360min\":2,\"trans_date_trans_time\":\"2012-01-29 03:33:21\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=150, serialized_header_size=-1)\n```\n:::\n:::\n\n\n:::\n\n## Next steps\n\nWoohoo, great job! Now that you've connected to Flink and learned the basics, you\ncan query your own data. See the rest of the Ibis documentation or [Flink\ndocumentation](https://nightlies.apache.org/flink/flink-docs-stable/). You can\n[open an issue](https://github.com/ibis-project/ibis/issues/new/choose) if you run\ninto one!\n\n",
+    "markdown": "---\ntitle: 'A real-life use case: fraud detection'\n---\n\n\n\nImagine you’re a data scientist who works at a large bank. You have been tasked\nwith one of the most challenging problems in banking today: identifying\nfraudulent transactions. The bank receives transaction details from its credit\ncard customers in a Kafka topic, which include information about the transaction\ndate and time, transaction amount, transaction location, merchant, category of\npurchase, and so on. Given the nature of the data, you want to use Apache Flink\nfor its stream processing capabilities and to develop machine learning features\nthat can be used to identify fraud.\n\n\n### Prerequisites\n\n* Docker Compose: This tutorial uses Docker Compose to manage an Apache Kafka\nenvironment (including sample data generation) and a Flink cluster (for remote\nexecution). You can [download and install Docker Compose from the official\nwebsite](https://docs.docker.com/compose/install/).\n* JDK 11 release: Flink requires Java 11.\n* Python 3.9 or 3.10.\n* Follow [the setup tutorial](0_setup.qmd) to install the Flink backend for Ibis.\n* Clone the [example repository](https://github.com/ibis-project/ibis-flink-tutorial).\n\n\n### Spinning up the services using Docker Compose\n\nFrom your project directory, run `docker compose up -d` to create Kafka topics,\ngenerate sample data, and launch a Flink cluster in the background.\n\n\n\n::: {.callout-tip}\nRunning `docker compose up` with the `-d` flag runs it in\ndetached mode, where containers are run in the background. While this frees up\nthe terminal for you to run other commands, it also hides error messages.\n\nDepending on whether the container images are already available locally, setting\nup the containers may take anywhere from 10 seconds to a minute. If it's your\nfirst time running this command, it's best to run it in the foreground so that\nyou can monitor the progress of setup.\n:::\n\nThis should set up a `transaction` topic in the Kafka cluster that contains\nmessages that look like the following:\n\n```\n{ \"trans_date_trans_time\": \"2012-02-23 00:10:01\", \"cc_num\":\n4428780000000000000, \"merchant\": \"fraud_Olson, Becker and Koch\", \"category\":\n\"gas_transport\", \"amt\": 82.55, \"first\": \"Richard\", \"last\": \"Waters\", \"zipcode\":\n\"53186\", \"dob\": \"1/2/46\", \"trans_num\": \"dbf31d83eebdfe96d2fa213df2043586\",\n\"is_fraud\": 0, \"user_id\": 7109464218691269943 }\n```\n\n::: {.callout-warning}\nDo not proceed to the next section until messages are\nflowing into the `transaction` topic!\n:::\n\n\n### Connect to a Flink environment session\n\nWe can connect to a Flink environment session by creating a\n`pyflink.table.TableEnvironment` and passing this to Flink backend’s `connect`\nmethod. For this tutorial, we are going to use Flink in streaming mode.\n\n::: {#e3f9188a .cell execution_count=2}\n``` {.python .cell-code}\nfrom pyflink.table import EnvironmentSettings, TableEnvironment\n\nimport ibis\n\nenv_settings = EnvironmentSettings.in_streaming_mode()\ntable_env = TableEnvironment.create(env_settings)\ntable_env.get_config().set(\"parallelism.default\", \"1\")\nconnection = ibis.flink.connect(table_env)\n```\n:::\n\n\nThe Kafka connector isn’t part of the binary distribution. In order to connect\nto a Kafka source/sink, we need to [download the JAR\nfile](https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/3.0.2-1.18/flink-sql-connector-kafka-3.0.2-1.18.jar)\nand manually add it into the classpath:\n\n::: {#4d3509b6 .cell execution_count=3}\n``` {.python .cell-code}\n!wget -N https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/3.0.2-1.18/flink-sql-connector-kafka-3.0.2-1.18.jar\nconnection.raw_sql(\"ADD JAR './flink-sql-connector-kafka-3.0.2-1.18.jar'\")\n```\n:::\n\n\nNow that we’ve set up the Flink table environment, we’re ready to connect to\ndata!\n\n\n### Connect to a data source\n\nIn order to experiment with the data in the Kafka topic and create\ntransformations on top of it, we need to first define and connect to the data\nsource.\n\nWhile we’re dealing with a continuous stream of data here, Flink and Ibis\nabstract differences in the underlying implementation between tables and\nstreams, so that, conceptually, we can simply treat our Kafka topic as a table.\n\nTo connect to our `transaction` Kafka topic, we need to provide a table name,\nschema of the data, and [connector\nconfigurations](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/connectors/table/kafka/#connector-options).\nThe schema of the data must contain a subset of the fields in the actual Kafka\ntopic. Because this is a streaming job, we also want to define a watermark\nstrategy for the data source by specifying the timestamp column (`time_col`) and\na time duration during which late events are accepted (`allowed_delay`). (If you\nare not already familiar with these concepts, you can check out [Flink’s\ndocumentation](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/datastream/operators/windows/)\nfor more details.) Note that [Flink requires the timestamp column to be of data\ntype\nTIMESTAMP(3)](https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/create/#watermark).\n\n::: {#38ca7a15 .cell execution_count=4}\n``` {.python .cell-code}\nimport ibis\nimport ibis.expr.datatypes as dt\nimport ibis.expr.schema as sch\n\nsource_schema = sch.Schema(\n    {\n        \"user_id\": dt.int64,\n        \"trans_date_trans_time\": dt.timestamp(scale=3),\n        \"cc_num\": dt.int64,\n        \"amt\": dt.float64,\n        \"trans_num\": dt.str,\n        \"merchant\": dt.str,\n        \"category\": dt.str,\n        \"is_fraud\": dt.int32,\n        \"first\": dt.str,\n        \"last\": dt.str,\n        \"dob\": dt.str,\n        \"zipcode\": dt.str,\n    }\n)\n\n# Configure the source table with Kafka connector properties.\nsource_configs = {\n    \"connector\": \"kafka\",\n    \"topic\": \"transaction\",\n    \"properties.bootstrap.servers\": \"localhost:9092\",\n    \"properties.group.id\": \"consumer_group_0\",\n    \"scan.startup.mode\": \"earliest-offset\",\n    \"format\": \"json\",\n}\n\n# Create the source table using the defined schema, Kafka connector properties,\n# and set watermarking for real-time processing with a 15-second allowed\n# lateness.\nsource_table = connection.create_table(\n    \"transaction\",\n    schema=source_schema,\n    tbl_properties=source_configs,\n    watermark=ibis.watermark(\n        time_col=\"trans_date_trans_time\", allowed_delay=ibis.interval(seconds=15)\n    ),\n)\n```\n:::\n\n\nWe’re ready to write some transformations!\n\n\n### Create transformations\n\nWhich signs could be indicative of suspected fraud in a credit card? Oftentimes,\nwe’re looking for abnormalities in user behaviors, for example, an excessively\nlarge transaction amount, unusually frequent transactions during a short period\nof time, etc. Based on this, the average transaction amount and the total\ntransaction count over the past five hours may be useful features. Let’s write\nout each of these using Ibis API:\n\n::: {#17a8cbb9 .cell execution_count=5}\n``` {.python .cell-code}\nuser_trans_amt_last_360m_agg = source_table[\n    source_table.user_id,\n    # Calculate the average transaction amount over the past six hours\n    source_table.amt.mean()\n    .over(\n        ibis.window(\n            group_by=source_table.user_id,\n            order_by=source_table.trans_date_trans_time,\n            range=(-ibis.interval(minutes=360), 0),\n        )\n    )\n    .name(\"user_mean_trans_amt_last_360min\"),\n    # Calculate the total transaction count over the past six hours\n    source_table.amt.count()\n    .over(\n        ibis.window(\n            group_by=source_table.user_id,\n            order_by=source_table.trans_date_trans_time,\n            range=(-ibis.interval(minutes=360), 0),\n        )\n    )\n    .name(\"user_trans_count_last_360min\"),\n    source_table.trans_date_trans_time,\n]\n```\n:::\n\n\n`over()` creates an [over\naggregation](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/sql/queries/over-agg/)\nin Flink, which computes an aggregated value for every input row. More\nspecifically, this means that an aggregation result is computed and emitted for\nevery new record flowing into the upstream Kafka topic.\n\nThe issue with over aggregation is that, if there is no new transaction for a\nspecific user during a time window, there would be no aggregation result written\nto the sink. In other words, the user would never show up in the result table if\nthey never made a transaction.\n\nAlternatively, we can compute aggregations using [Flink’s windowing table-valued\nfunctions](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/sql/queries/window-tvf/).\nThis allows more flexibility in defining windows and when results are computed\nand emitted into the sink. There are three types of windowing TVFs available in\nFlink: tumble, hop, and cumulate. Let’s define the same features with tumble\nwindows:\n\n::: {#bc66e887 .cell execution_count=6}\n``` {.python .cell-code}\nwindowed_stream = source_table.window_by(\n    time_col=source_table.trans_date_trans_time,\n).tumble(window_size=ibis.interval(minutes=360))\n\nuser_trans_amt_last_360m_agg_windowed_stream = windowed_stream.group_by(\n    [\"window_start\", \"window_end\", \"user_id\"]\n).agg(\n    user_mean_trans_amt_last_360min=windowed_stream.amt.mean(),\n    user_trans_count_last_360min=windowed_stream.amt.count(),\n)\n```\n:::\n\n\n### Connect to a data sink\n\nWe’re creating streaming jobs to continuously process upstream data, which could\nbe infinite. Therefore, we want to have the job continuously running and write\nresults into a data sink. Here, we’re simply going to write results into a\nseparate Kafka topic named `user_trans_amt_last_360min` for convenient\ndownstream processing.\n\nWe can define a data sink in virtually the same exact way in which we defined\nour data source:\n\n::: {#df490eeb .cell execution_count=7}\n``` {.python .cell-code}\nsink_schema = sch.Schema(\n    {\n        \"user_id\": dt.int64,\n        \"user_mean_trans_amt_last_360min\": dt.float64,\n        \"user_trans_count_last_360min\": dt.int64,\n        \"trans_date_trans_time\": dt.timestamp(scale=3),\n    }\n)\n\n# Configure the sink table with Kafka connector properties for writing results.\nsink_configs = {\n    \"connector\": \"kafka\",\n    \"topic\": \"user_trans_amt_last_360min\",\n    \"properties.bootstrap.servers\": \"localhost:9092\",\n    \"format\": \"json\",\n}\n\nsink_table = connection.create_table(\n    \"user_trans_amt_last_360min\",\n    schema=sink_schema,\n    tbl_properties=sink_configs,\n)\n```\n:::\n\n\nThe last step is to connect the pieces and actually write our query results into\nthe sink table that we had just created:\n\n::: {#8115681d .cell execution_count=8}\n``` {.python .cell-code}\nconnection.insert(\"user_trans_amt_last_360min\",\nuser_trans_amt_last_360m_agg)\n```\n\n::: {.cell-output .cell-output-display execution_count=8}\n```\n<pyflink.table.table_result.TableResult at 0x16ba87820>\n```\n:::\n:::\n\n\nThis step is exactly the same for windowing TVFs:\n\n::: {#44a37ca1 .cell execution_count=9}\n``` {.python .cell-code}\nsink_schema = sch.Schema(\n    {\n        \"window_start\": dt.timestamp(scale=3),\n        \"window_end\": dt.timestamp(scale=3),\n        \"user_id\": dt.int64,\n        \"user_mean_trans_amt_last_360min\": dt.float64,\n        \"user_trans_count_last_360min\": dt.int64,\n    }\n)\n\n# Configure the sink table with Kafka connector properties for writing results.\nsink_configs = {\n    \"connector\": \"kafka\",\n    \"topic\": \"user_trans_amt_last_360min_windowed\",\n    \"properties.bootstrap.servers\": \"localhost:9092\",\n    \"format\": \"json\",\n}\n\nsink_table = connection.create_table(\n    \"user_trans_amt_last_360min_windowed\",\n    schema=sink_schema,\n    tbl_properties=sink_configs,\n)\n\nconnection.insert(\n    \"user_trans_amt_last_360min_windowed\", user_trans_amt_last_360m_agg_windowed_stream\n)\n```\n\n::: {.cell-output .cell-output-display execution_count=9}\n```\n<pyflink.table.table_result.TableResult at 0x16bab58d0>\n```\n:::\n:::\n\n\n### Expected output\n\nNow, if everything is working correctly, you should expect to see results being\nstreamed into the Kafka topic!\n\n::: {.callout-tip}\nYou can inspect the Kafka topic using the Python Kafka client\nif you have it installed or via console Kafka consumer:\n\n::: {#462ccf08 .cell execution_count=10}\n``` {.python .cell-code}\nfrom kafka import KafkaConsumer\n\nconsumer = KafkaConsumer(\n    \"user_trans_amt_last_360min\"\n)  # or \"user_trans_amt_last_360min_windowed\"\nfor _, msg in zip(range(10), consumer):\n    print(msg)\n```\n\n::: {.cell-output .cell-output-stdout}\n```\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=45959, timestamp=1711571328876, timestamp_type=0, key=None, value=b'{\"user_id\":6405112720489830090,\"user_mean_trans_amt_last_360min\":64.675,\"user_trans_count_last_360min\":2,\"trans_date_trans_time\":\"2012-01-28 06:14:23\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=151, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=45960, timestamp=1711571328876, timestamp_type=0, key=None, value=b'{\"user_id\":-4272782897993296960,\"user_mean_trans_amt_last_360min\":61.34000000000002,\"user_trans_count_last_360min\":2,\"trans_date_trans_time\":\"2012-01-28 06:16:24\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=163, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=45961, timestamp=1711571328876, timestamp_type=0, key=None, value=b'{\"user_id\":-8404995308417450558,\"user_mean_trans_amt_last_360min\":47.17000000000002,\"user_trans_count_last_360min\":1,\"trans_date_trans_time\":\"2012-01-28 06:17:39\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=163, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=45962, timestamp=1711571328876, timestamp_type=0, key=None, value=b'{\"user_id\":-6232014680614930868,\"user_mean_trans_amt_last_360min\":131.91,\"user_trans_count_last_360min\":1,\"trans_date_trans_time\":\"2012-01-28 06:21:00\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=152, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=45963, timestamp=1711571328876, timestamp_type=0, key=None, value=b'{\"user_id\":1091996638939120426,\"user_mean_trans_amt_last_360min\":39.93000000000001,\"user_trans_count_last_360min\":2,\"trans_date_trans_time\":\"2012-01-28 06:21:06\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=162, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=45964, timestamp=1711571328876, timestamp_type=0, key=None, value=b'{\"user_id\":8490370068944420319,\"user_mean_trans_amt_last_360min\":179.13999999999996,\"user_trans_count_last_360min\":1,\"trans_date_trans_time\":\"2012-01-28 06:21:12\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=163, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=45965, timestamp=1711571328876, timestamp_type=0, key=None, value=b'{\"user_id\":969103286379660111,\"user_mean_trans_amt_last_360min\":85.7200000000001,\"user_trans_count_last_360min\":1,\"trans_date_trans_time\":\"2012-01-28 06:22:26\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=160, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=45966, timestamp=1711571328877, timestamp_type=0, key=None, value=b'{\"user_id\":-8414965219550676411,\"user_mean_trans_amt_last_360min\":6.029999999999972,\"user_trans_count_last_360min\":1,\"trans_date_trans_time\":\"2012-01-28 06:23:31\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=163, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=45967, timestamp=1711571328877, timestamp_type=0, key=None, value=b'{\"user_id\":-2480376382981778282,\"user_mean_trans_amt_last_360min\":8.270000000000007,\"user_trans_count_last_360min\":1,\"trans_date_trans_time\":\"2012-01-28 06:27:12\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=163, serialized_header_size=-1)\nConsumerRecord(topic='user_trans_amt_last_360min', partition=0, offset=45968, timestamp=1711571328877, timestamp_type=0, key=None, value=b'{\"user_id\":-5007797165712520135,\"user_mean_trans_amt_last_360min\":199.33,\"user_trans_count_last_360min\":1,\"trans_date_trans_time\":\"2012-01-28 06:28:10\"}', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=152, serialized_header_size=-1)\n```\n:::\n:::\n\n\n:::\n\n## Next steps\n\nWoohoo, great job! Now that you've connected to Flink and learned the basics, you\ncan query your own data. See the rest of the Ibis documentation or [Flink\ndocumentation](https://nightlies.apache.org/flink/flink-docs-stable/). You can\n[open an issue](https://github.com/ibis-project/ibis/issues/new/choose) if you run\ninto one!\n\n",
     "supporting": [
       "1_single_feature_files"
     ],
diff --git a/docs/contribute/02_workflow.qmd b/docs/contribute/02_workflow.qmd
index e8ae9d5d5..99c713af9 100644
--- a/docs/contribute/02_workflow.qmd
+++ b/docs/contribute/02_workflow.qmd
@@ -94,6 +94,7 @@ all ordinary and edge cases.
 
 Pytest markers can be used to assert that a test should fail or raise a specific error.
 We use a number of pytest markers in ibis:
+
 - `pytest.mark.notimpl`: the backend can do a thing, we haven't mapped the op
 - `pytest.mark.notyet`: the backend cannot do a thing, but might in the future
 - `pytest.mark.never`: the backend will never support this / pass this test (common example
diff --git a/docs/tutorials/open-source-software/apache-flink/1_single_feature.qmd b/docs/tutorials/open-source-software/apache-flink/1_single_feature.qmd
index 1f64e5f7f..86747c5d8 100644
--- a/docs/tutorials/open-source-software/apache-flink/1_single_feature.qmd
+++ b/docs/tutorials/open-source-software/apache-flink/1_single_feature.qmd
@@ -94,7 +94,7 @@ file](https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-
 and manually add it into the classpath:
 
 ```{python}
-# | include: false
+#| output: false
 !wget -N https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/3.0.2-1.18/flink-sql-connector-kafka-3.0.2-1.18.jar
 connection.raw_sql("ADD JAR './flink-sql-connector-kafka-3.0.2-1.18.jar'")
 ```
diff --git a/ibis/expr/api.py b/ibis/expr/api.py
index 1bd12ea86..d9ca23f98 100644
--- a/ibis/expr/api.py
+++ b/ibis/expr/api.py
@@ -313,15 +313,25 @@ _table_names = (f"unbound_table_{i:d}" for i in itertools.count())
 def table(
     schema: SchemaLike | None = None,
     name: str | None = None,
+    catalog: str | None = None,
+    database: str | None = None,
 ) -> ir.Table:
     """Create a table literal or an abstract table without data.
 
+    Ibis uses the word database to refer to a collection of tables, and the word
+    catalog to refer to a collection of databases. You can use a combination of
+    `catalog` and `database` to specify a hierarchical location for table.
+
     Parameters
     ----------
     schema
         A schema for the table
     name
         Name for the table. One is generated if this value is `None`.
+    catalog
+        A collection of database.
+    database
+        A collection of tables. Required if catalog is not `None`.
 
     Returns
     -------
@@ -340,13 +350,32 @@ def table(
       a int64
       b string
 
+
+    Create a table with no data backing it in a specific location
+
+    >>> import ibis
+    >>> ibis.options.interactive = False
+    >>> t = ibis.table(schema=dict(a="int"), name="t", catalog="cat", database="db")
+    >>> t
+    UnboundTable: cat.db.t
+      a int64
     """
     if name is None:
         if isinstance(schema, type):
             name = schema.__name__
         else:
             name = next(_table_names)
-    return ops.UnboundTable(name=name, schema=schema).to_expr()
+    if catalog is not None and database is None:
+        raise ValueError(
+            "A catalog-only namespace is invalid in Ibis, "
+            "please specify a database as well."
+        )
+
+    return ops.UnboundTable(
+        name=name,
+        schema=schema,
+        namespace=ops.Namespace(catalog=catalog, database=database),
+    ).to_expr()
 
 
 def memtable(
diff --git a/ibis/expr/format.py b/ibis/expr/format.py
index 2975de37d..2c5186bcd 100644
--- a/ibis/expr/format.py
+++ b/ibis/expr/format.py
@@ -244,6 +244,14 @@ def _physical_table(op, name, **kwargs):
     return f"{op.__class__.__name__}: {name}\n{schema}"
 
 
+@fmt.register(ops.UnboundTable)
+@fmt.register(ops.DatabaseTable)
+def _unbound_table(op, name, **kwargs):
+    schema = render_schema(op.schema, indent_level=1)
+    name = ".".join(filter(None, op.namespace.args + (name,)))
+    return f"{op.__class__.__name__}: {name}\n{schema}"
+
+
 @fmt.register(ops.InMemoryTable)
 def _in_memory_table(op, data, **kwargs):
     import rich.pretty
diff --git a/ibis/expr/operations/relations.py b/ibis/expr/operations/relations.py
index 215bcd142..fd6f73f76 100644
--- a/ibis/expr/operations/relations.py
+++ b/ibis/expr/operations/relations.py
@@ -313,8 +313,8 @@ class PhysicalTable(Relation):
 
 @public
 class Namespace(Concrete):
-    database: Optional[str] = None
     catalog: Optional[str] = None
+    database: Optional[str] = None
 
 
 @public
diff --git a/ibis/expr/types/logical.py b/ibis/expr/types/logical.py
index 3fcc4d745..a0c289df3 100644
--- a/ibis/expr/types/logical.py
+++ b/ibis/expr/types/logical.py
@@ -229,9 +229,32 @@ class BooleanValue(NumericValue):
         """
         return self.negate()
 
-    @staticmethod
-    def __negate_op__():
-        return ops.Not
+    def negate(self) -> BooleanValue:
+        """Negate a boolean expression.
+
+        Returns
+        -------
+        BooleanValue
+            A boolean value expression
+
+        Examples
+        --------
+        >>> import ibis
+        >>> ibis.options.interactive = True
+        >>> t = ibis.memtable({"values": [True, False, False, None]})
+        >>> t.values.negate()
+        ┏━━━━━━━━━━━━━┓
+        ┃ Not(values) ┃
+        ┡━━━━━━━━━━━━━┩
+        │ boolean     │
+        ├─────────────┤
+        │ False       │
+        │ True        │
+        │ True        │
+        │ NULL        │
+        └─────────────┘
+        """
+        return ops.Not(self).to_expr()
 
 
 @public
diff --git a/ibis/expr/types/numeric.py b/ibis/expr/types/numeric.py
index 27802600a..56d5c2dea 100644
--- a/ibis/expr/types/numeric.py
+++ b/ibis/expr/types/numeric.py
@@ -19,11 +19,6 @@ if TYPE_CHECKING:
 
 @public
 class NumericValue(Value):
-    @staticmethod
-    def __negate_op__():
-        # TODO(kszucs): do we need this?
-        return ops.Negate
-
     def negate(self) -> NumericValue:
         """Negate a numeric expression.
 
@@ -48,14 +43,7 @@ class NumericValue(Value):
         │             -1 │
         └────────────────┘
         """
-        op = self.op()
-        try:
-            result = op.negate()
-        except AttributeError:
-            op_class = self.__negate_op__()
-            result = op_class(self)
-
-        return result.to_expr()
+        return ops.Negate(self).to_expr()
 
     def __neg__(self) -> NumericValue:
         """Negate `self`.
diff --git a/ibis/expr/types/temporal.py b/ibis/expr/types/temporal.py
index c43fe6780..120f59b8e 100644
--- a/ibis/expr/types/temporal.py
+++ b/ibis/expr/types/temporal.py
@@ -939,20 +939,10 @@ class IntervalValue(Value):
         IntervalValue
             A negated interval value expression
         """
-        op = self.op()
-        if hasattr(op, "negate"):
-            result = op.negate()
-        else:
-            result = ops.Negate(self)
-
-        return result.to_expr()
+        return ops.Negate(self).to_expr()
 
     __neg__ = negate
 
-    @staticmethod
-    def __negate_op__():
-        return ops.Negate
-
 
 @public
 class IntervalScalar(Scalar, IntervalValue):


Test patch:
diff --git a/ibis/expr/tests/snapshots/test_format/test_format_unbound_table_namespace/repr.txt b/ibis/expr/tests/snapshots/test_format/test_format_unbound_table_namespace/repr.txt
new file mode 100644
index 000000000..5dcb87c3b
--- /dev/null
+++ b/ibis/expr/tests/snapshots/test_format/test_format_unbound_table_namespace/repr.txt
@@ -0,0 +1,3 @@
+UnboundTable: bork
+  a int64
+  b int64
\ No newline at end of file
diff --git a/ibis/expr/tests/snapshots/test_format/test_format_unbound_table_namespace/reprcatdb.txt b/ibis/expr/tests/snapshots/test_format/test_format_unbound_table_namespace/reprcatdb.txt
new file mode 100644
index 000000000..d25f4eb51
--- /dev/null
+++ b/ibis/expr/tests/snapshots/test_format/test_format_unbound_table_namespace/reprcatdb.txt
@@ -0,0 +1,3 @@
+UnboundTable: ork.bork.bork
+  a int64
+  b int64
\ No newline at end of file
diff --git a/ibis/expr/tests/snapshots/test_format/test_format_unbound_table_namespace/reprdb.txt b/ibis/expr/tests/snapshots/test_format/test_format_unbound_table_namespace/reprdb.txt
new file mode 100644
index 000000000..b6fe4c9be
--- /dev/null
+++ b/ibis/expr/tests/snapshots/test_format/test_format_unbound_table_namespace/reprdb.txt
@@ -0,0 +1,3 @@
+UnboundTable: bork.bork
+  a int64
+  b int64
\ No newline at end of file
diff --git a/ibis/expr/tests/test_api.py b/ibis/expr/tests/test_api.py
index edb5adbbe..975e65121 100644
--- a/ibis/expr/tests/test_api.py
+++ b/ibis/expr/tests/test_api.py
@@ -156,3 +156,18 @@ def test_nested_name_property():
         x = x + 1
 
     assert x.op().name.count("Add") == n
+
+
+def test_unbound_table_namespace():
+    t = ibis.table(name="bork", schema=(("a", "int"), ("b", "int")), database="bork")
+
+    assert t.op().namespace == ops.Namespace(database="bork")
+
+    t = ibis.table(
+        name="bork", schema=(("a", "int"), ("b", "int")), database="bork", catalog="ork"
+    )
+
+    assert t.op().namespace == ops.Namespace(catalog="ork", database="bork")
+
+    with pytest.raises(ValueError, match="A catalog-only namespace is invalid in Ibis"):
+        ibis.table(name="bork", schema=(("a", "int"), ("b", "int")), catalog="bork")
diff --git a/ibis/expr/tests/test_format.py b/ibis/expr/tests/test_format.py
index d8ab0ecc9..63ebda4f1 100644
--- a/ibis/expr/tests/test_format.py
+++ b/ibis/expr/tests/test_format.py
@@ -379,6 +379,25 @@ def test_format_in_memory_table(snapshot):
     snapshot.assert_match(result, "repr.txt")
 
 
+def test_format_unbound_table_namespace(snapshot):
+    t = ibis.table(name="bork", schema=(("a", "int"), ("b", "int")))
+
+    result = fmt(t)
+    snapshot.assert_match(result, "repr.txt")
+
+    t = ibis.table(name="bork", schema=(("a", "int"), ("b", "int")), database="bork")
+
+    result = fmt(t)
+    snapshot.assert_match(result, "reprdb.txt")
+
+    t = ibis.table(
+        name="bork", schema=(("a", "int"), ("b", "int")), catalog="ork", database="bork"
+    )
+
+    result = fmt(t)
+    snapshot.assert_match(result, "reprcatdb.txt")
+
+
 def test_format_new_relational_operation(alltypes, snapshot):
     class MyRelation(ops.Relation):
         parent: ops.Relation