Skip to content

[SQL] Connector transport/format validation in the SQL compiler #6538

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mihaibudiu merged 1 commit into from
Jun 26, 2026
Merged

[SQL] Connector transport/format validation in the SQL compiler #6538

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
262 changes: 262 additions & 0 deletions crates/adapters/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,3 +51,265 @@ $ cargo run --example server --features="with-kafka server"
```

Then open a web browser and open the following URL: `http://localhost:8080`

## Connector configuration validation

The SQL compiler validates connector configurations at compile time, reporting
warnings for unknown fields and invalid values before the pipeline ever runs.
This section explains how the validation works and what to do when you add or
change a Rust connector config struct.

### Architecture

The validation lives in the SQL compiler:

```
sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/
├── ConnectorValidator.java — dispatch: routes format/transport names to POJO classes
├── ConfigReporter.java — emits positioned warnings via JSON Pointer lookup
├── IValidateConfig.java — interface: boolean validate(ConfigReporter)
└── config/ — one POJO class (and helper enums) per Rust config struct
├── CsvParserConfig.java
├── KafkaInputConfig.java
└── ...
```

`ConnectorValidator.validateFormatConfig()` and `validateTransportConfig()` are called
once per connector during SQL compilation. Each method looks at the transport/format
name and deserializes the JSON `config` block into the matching POJO class. Unknown
fields produce a warning pointing at the exact source location of the typo. After
deserialization the POJO's `validate()` method checks cross-field constraints.

By default every unknown field is rejected. A POJO opts out of that check for specific
keys by declaring a `@JsonAnySetter` method:

```java
/** Absorbs flattened extra keys so they are not rejected as unknown. */
@JsonAnySetter
public void setExtra(String key, Object value) {}
```

Jackson calls the method for unknown keys instead of failing. Since the
method body is empty the key is silently absorbed. This is used for configs that have
a `#[serde(flatten)] HashMap<String, String>` in Rust (e.g. Kafka, Delta Table, Iceberg),
where any string key is valid and should not be reported as an error.

### Currently validated connectors

#### Format (input)
| Name | POJO class |
|------------|-----------------------|
| `avro` | `AvroParserConfig` |
| `csv` | `CsvParserConfig` |
| `json` | `JsonParserConfig` |
| `parquet` | `ParquetParserConfig` |
| `raw` | `RawParserConfig` |

#### Format (output)
| Name | POJO class |
|------------|------------------------|
| `avro` | `AvroEncoderConfig` |
| `csv` | `CsvEncoderConfig` |
| `json` | `JsonEncoderConfig` |
| `parquet` | `ParquetEncoderConfig` |

#### Transport (input)
| Name | POJO class |
|----------------------|---------------------------|
| `clock` | `ClockConfig` |
| `delta_table_input` | `DeltaTableReaderConfig` |
| `file_input` | `FileInputConfig` |
| `iceberg_input` | `IcebergReaderConfig` |
| `kafka_input` | `KafkaInputConfig` |
| `nats_input` | `NatsInputConfig` |
| `postgres_input` | `PostgresReaderConfig` |
| `postgres_cdc_input` | `PostgresCdcReaderConfig` |
| `pub_sub_input` | `PubSubInputConfig` |
| `s3_input` | `S3InputConfig` |
| `url_input` | `UrlInputConfig` |

#### Transport (output)
| Name | POJO class |
|----------------------|--------------------------|
| `delta_table_output` | `DeltaTableWriterConfig` |
| `dynamodb_output` | `DynamoDBWriterConfig` |
| `file_output` | `FileOutputConfig` |
| `http_output` | `HttpOutputConfig` |
| `kafka_output` | `KafkaOutputConfig` |
| `postgres_output` | `PostgresWriterConfig` |
| `redis_output` | `RedisOutputConfig` |

Not validated by the SQL compiler: `datagen`, `nexmark`.

Note: flattened string-map configs (e.g., `kafka_options`) cannot be
validated to detect typos.

#### Data-carrying enum variants

Some Rust enums have variants that carry data, e.g.:

```rust
pub enum DeliverPolicy {
All,
Last,
ByStartSequence { start_sequence: u64 },
ByStartTime { start_time: OffsetDateTime },
}
```

Rust's default serde serialization is **externally tagged**: unit variants become
plain strings (`"All"`), struct variants become single-key objects
(`{"ByStartSequence": {"start_sequence": 5}}`). Jackson has no direct equivalent
without a custom deserializer. The recommended approach is to declare the field
as `JsonNode` in the Java POJO and skip content validation:

```java
/** Required. Unit variants ("All", "Last", ...) and struct variants
* ({"ByStartSequence": {...}}) are accepted as-is. */
@Nullable
@JsonProperty("deliver_policy")
public JsonNode deliverPolicy = null;
```

Presence can still be checked in `validate()` via
`deliverPolicy == null || deliverPolicy.isNull()`.

### How to update the Java code

The Java compiler validates connector configuration, but only gives
warnings. The ultimate real validation is done by the connector code
itself, in Rust, at runtime.

#### When you add a field to an existing Rust struct

Find the matching POJO class in `config/` and add the field following the

@mythical-fred mythical-fred Jun 25, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Worth documenting the fail-open contract explicitly: (a) unknown transport/format names are silently skipped by the SQL compiler, (b) Jackson deserialization errors surface as warnings not errors, (c) @JsonAnySetter keys are not validated at all. Future contributors will otherwise assume "SQL compiler accepted it → it works." Also: this README does an excellent job of explaining mechanics, but nowhere says "the runtime is still the source of truth — don't remove Rust-side validation when you add a Java mirror." Worth a one-liner since the whole point of the doc is to guide future connector additions.

mapping rules below.

#### When you add a new Rust config struct

1. **Create a POJO class** in `config/` implementing `IValidateConfig`:

```java
package org.dbsp.sqlCompiler.compiler.frontend.connectors.config;

import org.dbsp.sqlCompiler.compiler.frontend.connectors.ConfigReporter;
import org.dbsp.sqlCompiler.compiler.frontend.connectors.IValidateConfig;
import com.fasterxml.jackson.annotation.JsonProperty;
import javax.annotation.Nullable;

public class MyTransportConfig implements IValidateConfig {
@JsonProperty("some_field")
public String someField = "";

@Override
public boolean validate(ConfigReporter reporter) {
boolean ok = true;
if (someField.isBlank()) {
reporter.warnPath("some_field", "Invalid configuration",
"required field \"some_field\" is missing or empty");
ok = false;
}
return ok;
}
}
```

2. **Wire it** in `ConnectorValidator.validateTransportConfig()` (or
`validateFormatConfig()` for formats):

```java
case "my_transport_input":
validateConfig(transportConfig, outerJson, configPointer,
outerStart, MyTransportConfig.class, reporter);
break;
```

3. **Add tests** in `ConnectorTests.java` using the `tableConnectorTest()` /
`viewConnectorTest()` helpers.

#### When you rename or remove a field

Update the `@JsonProperty` annotation (or remove the field from the POJO).
If the old name should still be accepted as a fallback, add `@JsonAlias("old_name")`.

#### When you add a new enum variant

Add the variant to the Java enum with a matching `@JsonProperty`. If the Rust
variant has `#[serde(skip)]`, omit it from the Java enum entirely.

### Rust-to-Java field mapping

| Rust | Java |
|---------------------------------------------------|-------------------------------------------------------------------|
| `field: String` | `@JsonProperty("field") public String field = "";` |
| `field: Option<String>` | `@Nullable @JsonProperty("field") public String field = null;` |
| `flag: bool` (default false) | `@JsonProperty("flag") public boolean flag = false;` |
| `#[serde(default = "f")] n: u32` | `@JsonProperty("n") public int n = /* value of f() */;` |
| `#[serde(rename = "x")] field: T` | `@JsonProperty("x") public T field = ...;` |
| `#[serde(alias = "old")] field: T` | `@JsonAlias("old") @JsonProperty("new") public T field;` |
| `#[serde(flatten)] extra: HashMap<String, String>`| `@JsonAnySetter public void set(String k, Object v) {}` |
| `#[serde(flatten)] inner: InnerStruct` | Inline all fields of `InnerStruct` directly (see below) |
| enum variant `#[serde(rename = "x")] Foo` | `@JsonProperty("x") Foo,` |
| enum variant `#[serde(skip)] Bar` | Omit from the Java enum |

#### Java field initializers

Java field initializers in POJO classes serve **validation logic only** — they are
not used by the runtime, which has its own Rust defaults. Set an initializer only
when `validate()` depends on it:

- `String field = ""` — sentinel for a required field; `validate()` checks
`field.isBlank()` to detect omission.
- `int n = k` — when `validate()` checks a range or relationship involving `n` and
the Rust default `k` is the boundary (e.g. `n == 0` is invalid but the Rust
default is `1`, so the initializer must be `1` for the check to fire correctly on
omission).

Do **not** copy Rust defaults just for documentation — the Rust and Java values are
not kept in sync automatically, so a stale initializer is worse than no initializer.
For fields that `validate()` never inspects, omit the initializer or use the Java
natural default (`0`, `false`, `null`).

#### Flattened structs

Copy all fields from the nested Rust struct directly into the Java POJO class:

```rust
// Rust
pub struct OuterConfig {
pub name: String,
#[serde(flatten)]
pub tls: TlsConfig, // has fields ssl_ca_pem, verify_hostname, ...
}
```

```java
// Java — inline the TLS fields directly
public class OuterConfig implements IValidateConfig {
@JsonProperty("name") public String name = "";
// TLS fields (inlined from TlsConfig)
@Nullable @JsonProperty("ssl_ca_pem") public String sslCaPem = null;
@Nullable @JsonProperty("verify_hostname") public Boolean verifyHostname = null;
// ...
}
```

### Emitting warnings from `validate()`

```java
// Point at a specific field's key in the source
reporter.warnPath("field_name", "Invalid configuration", "message");

// Point at the whole config block (when the error isn't tied to one field)
reporter.warn("Invalid configuration", "message");
```

`warnPath` accepts a slash-separated JSON Pointer suffix relative to the config
block, e.g. `"field"` or `"nested/subfield"`.

### Running the tests

```bash
cd sql-to-dbsp-compiler/SQL-compiler
mvn test -Dtest=ConnectorTests
```
5 changes: 5 additions & 0 deletions ...rc/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/SqlToRelCompiler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,7 @@
import org.dbsp.sqlCompiler.compiler.frontend.statements.CreateAggregateStatement;
import org.dbsp.sqlCompiler.compiler.frontend.statements.CreateFunctionStatement;
import org.dbsp.sqlCompiler.compiler.frontend.statements.CreateIndexStatement;
import org.dbsp.sqlCompiler.compiler.frontend.connectors.ConnectorValidator;
import org.dbsp.sqlCompiler.compiler.frontend.statements.CreateTableStatement;
import org.dbsp.sqlCompiler.compiler.frontend.statements.CreateTypeStatement;
import org.dbsp.sqlCompiler.compiler.frontend.statements.CreateViewStatement;
Expand Down Expand Up @@ -2142,6 +2143,10 @@ void validateConnectorsProperty(CalciteObject ignored, boolean isTable, ProgramI
throw new CompilationError("Postprocessor field \"config\" must be a JSON object", pos);
}
}
ConnectorValidator.validateFormatConfig(connector, path, isTable, json,
value.getSourcePosition().start, this.errorReporter);
ConnectorValidator.validateTransportConfig(connector, path, isTable, json,
value.getSourcePosition().start, this.errorReporter);
}
} else {
var error = jsonNode.err();
Expand Down
65 changes: 65 additions & 0 deletions ...piler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/ConfigReporter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
package org.dbsp.sqlCompiler.compiler.frontend.connectors;

import org.dbsp.sqlCompiler.compiler.IErrorReporter;
import org.dbsp.sqlCompiler.compiler.errors.SourcePosition;
import org.dbsp.sqlCompiler.compiler.errors.SourcePositionRange;
import org.dbsp.util.Utilities;

/**
* A reporter that can resolve field names to source positions within a JSON config object.
*/
public final class ConfigReporter {
private final IErrorReporter reporter;
private final String outerJson;
private final String configJPath;
/** Start source position of the JSON document that is being validated. */
private final SourcePosition outerStart;

/**
* Creates a reporter that resolves field names to source positions within a JSON
* config object.
*
* @param reporter Underlying error reporter used to emit warnings.
* @param outerJson The raw connectors-array JSON string (the value of the SQL
* {@code 'connectors'} property). {@code configPointer} is a
* JSON Pointer path into this array that identifies the config block.
* @param configPointer JSON Pointer path to the config block within {@code outerJson}
* (e.g. {@code "/0/transport/config"}). Field-level warnings
* append a {@code /fieldName} suffix to this pointer to locate
* the exact key in the source.
* @param outerStart Absolute source position in the SQL program
* of {@code outerJson}'s first character,
* used to convert within-JSON byte offsets to source positions.
*/
public ConfigReporter(IErrorReporter reporter, String outerJson,
String configPointer, SourcePosition outerStart) {
this.reporter = reporter;
this.outerJson = outerJson;
this.configJPath = configPointer;
this.outerStart = outerStart;
}

/**
* Report a warning at the key token identified by
* {@code relativePath} within the config object. {@code relativePath} is a
* slash-separated relative JSON Pointer suffix (e.g. {@code "delimiter"} or
* {@code "quoting/delimiter"}); it is appended to the config pointer.
*/
public void warnPath(String relativePath, String category, String message) {
String pointer = this.configJPath + "/" + relativePath;
SourcePosition pos = Utilities.jsonPointerLocation(this.outerJson, pointer, false);
SourcePositionRange range = pos != null
? pos.relativeTo(this.outerStart).asRange()
: SourcePositionRange.INVALID;
this.reporter.reportWarning(range, category, message);
}

/** Report a warning that does not correspond to a single field (points at the config block). */
public void warn(String category, String message) {
SourcePosition pos = Utilities.jsonPointerLocation(this.outerJson, this.configJPath, true);
SourcePositionRange range = pos != null
? pos.relativeTo(this.outerStart).asRange()
: SourcePositionRange.INVALID;
this.reporter.reportWarning(range, category, message);
}
}
Loading