# Raw Format

:::note
This page describes configuration options specific to the raw data format.
See [top-level connector documentation](/connectors/) for general information
about configuring input and output connectors.
:::

The **raw** data format can be used to configure an input connector to pass
raw, unparsed, data buffers from the [transport connector](/connectors/sources) to the SQL program.
Normally the byte stream received from the transport connector is parsed using the specified data [format](/formats),
e.g., [JSON](/formats/json) or [Avro](/formats/avro). The parser converts the raw byte stream into a stream
of SQL table records.  In some cases it is preferable to pass the raw bytes to the SQL program unmodified and
perform all parsing in SQL. In particular this is useful for:

* **Custom data formats:** If the required data format is not supported by built-in parsers, raw data can be
  ingested into a SQL table and decoded using a [UDF](/sql/udf).

* **Parallelized parsing**: Parsing large volumes of incoming data can become a performance bottleneck.
  Normally parsing is performed by the input connector outside of the SQL runtime.
  Moving parsing into the SQL program enables it to leverage the parallelism in the Feldera
  SQL runtime.

The raw format is currently only supported for input connectors.

Note that the raw format supports only inserting records into the table and does not support deletions.
This limitation exists because the raw data stream lacks the metadata to differentiate between inserts and deletes.

## Configuring raw input connector

In order to ingest raw data:

1. Declare a table with a column of type `VARBINARY [NOT NUL]` or `VARCHAR [NOT NULL]`.  In the latter case, the connector
   will parse input data as a UTF-8 string and will fail to parse chunks that are not valid UTF-8.
   The table can contain additional columns, but all other columns in the table must be either nullable or
   have default values.

2. Configure an input connector with the format name `raw`.  The connector currently supports the following configuration
   options:

| Property                      | Type                        |Default  | Description                                              |
|-------------------------------|-----------------------------|---------|----------------------------------------------------------|
| `mode`                        | `"blob"` or `"lines"`       |`"blob"` | Ingestion mode:<ul><li>`blob` (default) - ingest the entire data chunk received from the transport connector as a single SQL row. For message-oriented transports, such as Kafka or Pub/Sub, an input chunk corresponds to a message. For file-based transports, e.g., the [URL](/connectors/sources/http-get) connector or the [S3](/connectors/sources/s3) connector, a chunk represents an entire file or object.</li><li>`lines` - split the input byte stream on the new line character (`\n`) and ingest each line as a separate SQL row.</li></ul>|
| `column_name`                 | string                      |         | Table column that will store the raw value. This setting is required if the table has more than 1 column.|

:::note

When using the `blob` mode with a file-oriented transport connector such as S3, the entire file or object is ingested
as a single record.  This can require a lot of memory when reading large files.

:::

### Example 1: Ingest raw data from URL

The following example shows a table with an input connector configured to ingest raw data from a URL line-by-line.

```sql
create table raw_table(
    data varchar
) with (
    'connectors' = '[{
        "format": {
            "name": "raw",
            "config": {
                "mode": "lines"
            }
        },
        "transport": {
            "name": "url_input",
            "config": {
                "path": "https://feldera-basics-tutorial.s3.amazonaws.com/part.json"
            }
        }
    }]'
);
```

### Example 2: Ingest Kafka payload and metadata.

This example ingests Kafka message values as raw strings into the `data` column, and Kafka
message metadata into other columns (see [Accessing Kafka metadata](/connectors/sources/kafka#metadata)).

```sql
CREATE TABLE raw_table (
    data VARCHAR,
    kafka_headers MAP<STRING, VARBINARY> DEFAULT CAST(CONNECTOR_METADATA()['kafka_headers'] as MAP<STRING, VARBINARY>),
    kafka_timestamp TIMESTAMP DEFAULT CAST(CONNECTOR_METADATA()['kafka_timestamp'] as TIMESTAMP),
    kafka_topic VARCHAR DEFAULT CAST(CONNECTOR_METADATA()['kafka_topic'] AS VARCHAR),
    kafka_offset BIGINT DEFAULT CAST(CONNECTOR_METADATA()['kafka_offset'] AS BIGINT),
    kafka_partition INT DEFAULT CAST(CONNECTOR_METADATA()['kafka_partition'] AS INT)
) WITH (
  'materialized' = 'true',
  'connectors' = '[
    {
      "name": "raw_data",
      "transport": {
          "name": "kafka_input",
          "config": {
              "topic": "my_topic",
              "start_from": "earliest",
              "bootstrap.servers": "localhost:19092",
              "include_headers": true,
              "include_topic": true,
              "include_offset": true,
              "include_partition": true,
              "include_timestamp": true
          }
      },
      "format": {
          "name": "raw",
          "config": {
              "column_name": "data"
          }
      }
  }]'
)
```

## Ingesting raw data via HTTP

You can also push raw data to a pipeline via [HTTP](/connectors/sources/http) by specifying
`format=json` and, optionally, `mode=lines` in the request URL:

### Example

Create a pipeline called `my_pipeline` with the following table declaration:

```sql
create table raw_table(
    data varchar
);
```

Insert two records with values `hello` and `world`:

```bash
curl -i -X 'POST' \
  http://127.0.0.1:8080/v0/pipelines/my_pipeline/ingress/raw_table?format=raw&mode=lines \
  -d 'hello
  world'
```
