diff --git a/crates/adapters/README.md b/crates/adapters/README.md index 16b6ad85841..1baba6270b9 100644 --- a/crates/adapters/README.md +++ b/crates/adapters/README.md @@ -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` 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 +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` | `@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`| `@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 +``` diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/SqlToRelCompiler.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/SqlToRelCompiler.java index c5d8985547b..77cbf11c38a 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/SqlToRelCompiler.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/SqlToRelCompiler.java @@ -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; @@ -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(); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/ConfigReporter.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/ConfigReporter.java new file mode 100644 index 00000000000..966ae9d1ece --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/ConfigReporter.java @@ -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); + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/ConnectorValidator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/ConnectorValidator.java new file mode 100644 index 00000000000..fb79cc68e22 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/ConnectorValidator.java @@ -0,0 +1,379 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.JsonMappingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.exc.InvalidFormatException; +import com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException; +import com.google.common.util.concurrent.UncheckedTimeoutException; +import org.dbsp.sqlCompiler.compiler.IErrorReporter; +import org.dbsp.sqlCompiler.compiler.errors.SourcePosition; +import org.dbsp.sqlCompiler.compiler.errors.SourcePositionRange; +import org.dbsp.sqlCompiler.compiler.frontend.connectors.config.*; +import org.dbsp.util.Utilities; + +import javax.annotation.Nullable; +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; + +/** Validation helpers for connector format and transport configuration objects. */ +public final class ConnectorValidator { + private ConnectorValidator() {} + + /** Returns the comma-separated list of serialized names for all constants of an enum type. */ + private static String enumValidValues(Class enumType) { + Object[] constants = enumType.getEnumConstants(); + Utilities.enforce(constants != null, () -> "enumType must be an enum class"); + ObjectMapper mapper = Utilities.deterministicObjectMapper(); + return Arrays.stream(constants) + .map(c -> mapper.convertValue(c, JsonNode.class).toString()) + .collect(Collectors.joining(", ")); + } + + /** + * Converts a Jackson deserialization path to a JSON Pointer suffix + * (e.g. {@code "/field1/field2"}). + */ + private static String pathToPointer(List path) { + return path.stream() + .map(ref -> ref.getFieldName() != null + ? "/" + ref.getFieldName() + : "/" + ref.getIndex()) + .collect(Collectors.joining()); + } + + /** + * Converts a Jackson deserialization path to a dot-separated field name + * (e.g. {@code "field1.field2"}). + */ + private static String pathToDotted(List path) { + return path.stream() + .map(ref -> ref.getFieldName() != null + ? ref.getFieldName() : "[" + ref.getIndex() + "]") + .collect(Collectors.joining(".")); + } + + /** Emits a warning pointing at the key identified by {@code pointer}. */ + private static void warnAt(String outerJson, String pointer, SourcePosition outerStart, + IErrorReporter reporter, String category, String message) { + SourcePosition pos = Utilities.jsonPointerLocation(outerJson, pointer, false); + if (pos == null) + pos = Utilities.jsonPointerLocation(outerJson, pointer, true); + SourcePositionRange range = pos != null + ? pos.relativeTo(outerStart).asRange() + : SourcePositionRange.INVALID; + reporter.reportWarning(range, category, message); + } + + /** + * Validates the {@code format.config} object of a single connector JSON node, + * emitting warnings for unknown fields or wrong-typed values. + * + * @param connector Parsed JSON object for one connector. + * @param connectorPath JSON Pointer prefix for this connector within {@code outerJson} + * (e.g. {@code "/0"}). + * @param isTable {@code true} for table (input) connectors, {@code false} for + * view (output) connectors. + * @param outerJson The complete connectors JSON string. + * @param outerStart Absolute source position of {@code outerJson}'s first character + * in the SQL program. + * @param reporter Error reporter used. + */ + public static void validateFormatConfig( + JsonNode connector, String connectorPath, + boolean isTable, String outerJson, SourcePosition outerStart, + IErrorReporter reporter) { + JsonNode format = connector.get("format"); + if (format == null) + return; + if (!format.isObject()) { + warnAt(outerJson, connectorPath + "/format", outerStart, reporter, + "Invalid connector", "\"format\" must be a JSON object"); + return; + } + JsonNode formatNameNode = format.get("name"); + if (formatNameNode == null) { + warnAt(outerJson, connectorPath + "/format", outerStart, reporter, + "Invalid connector", "\"format\" must have a \"name\" field"); + return; + } + if (!formatNameNode.isTextual()) { + warnAt(outerJson, connectorPath + "/format/name", outerStart, reporter, + "Invalid connector", "\"format.name\" must be a string"); + return; + } + JsonNode formatConfig = format.get("config"); + if (formatConfig == null) + return; + if (!formatConfig.isObject()) { + warnAt(outerJson, connectorPath + "/format/config", outerStart, reporter, + "Invalid connector", "\"format.config\" must be a JSON object"); + return; + } + String configPointer = connectorPath + "/format/config"; + String formatName = formatNameNode.asText(); + if (isTable) { + switch (formatName) { + case "avro": + validateConfig(formatConfig, outerJson, configPointer, + outerStart, AvroParserConfig.class, reporter); + break; + case "csv": + validateConfig(formatConfig, outerJson, configPointer, + outerStart, CsvParserConfig.class, reporter); + break; + case "json": + validateConfig(formatConfig, outerJson, configPointer, + outerStart, JsonParserConfig.class, reporter); + break; + case "parquet": + validateConfig(formatConfig, outerJson, configPointer, + outerStart, ParquetParserConfig.class, reporter); + break; + case "raw": + validateConfig(formatConfig, outerJson, configPointer, + outerStart, RawParserConfig.class, reporter); + break; + default: + warnAt(outerJson, connectorPath + "/format/name", outerStart, reporter, + "Unknown format", "\"format.name\" " + Utilities.doubleQuote(formatName, true) + " is not known"); + break; + } + } else { + switch (formatName) { + case "avro": + validateConfig(formatConfig, outerJson, configPointer, + outerStart, AvroEncoderConfig.class, reporter); + break; + case "csv": + validateConfig(formatConfig, outerJson, configPointer, + outerStart, CsvEncoderConfig.class, reporter); + break; + case "json": + validateConfig(formatConfig, outerJson, configPointer, + outerStart, JsonEncoderConfig.class, reporter); + break; + case "parquet": + validateConfig(formatConfig, outerJson, configPointer, + outerStart, ParquetEncoderConfig.class, reporter); + break; + default: + warnAt(outerJson, connectorPath + "/format/name", outerStart, reporter, + "Unknown format", "\"format.name\" " + Utilities.doubleQuote(formatName, true) + " is not known"); + break; + } + } + } + + /** + * Validates the {@code transport.config} object of a single connector JSON node, + * emitting warnings for unknown fields or wrong-typed values. + * + * @param connector Parsed JSON object for one connector. + * @param connectorPath JSON Pointer prefix for this connector within {@code outerJson}. + * @param isTable {@code true} for table (input) connectors, {@code false} for + * view (output) connectors. + * @param outerJson The complete connectors JSON string. + * @param outerStart Absolute source position of {@code outerJson}'s first character + * in the SQL program. + * @param reporter Error reporter. + */ + public static void validateTransportConfig( + JsonNode connector, String connectorPath, + boolean isTable, String outerJson, SourcePosition outerStart, + IErrorReporter reporter) { + JsonNode transport = connector.get("transport"); + if (transport == null) + return; + if (!transport.isObject()) { + warnAt(outerJson, connectorPath + "/transport", outerStart, reporter, + "Invalid connector", "\"transport\" must be a JSON object"); + return; + } + JsonNode transportNameNode = transport.get("name"); + if (transportNameNode == null) { + warnAt(outerJson, connectorPath + "/transport", outerStart, reporter, + "Invalid connector", "\"transport\" must have a \"name\" field"); + return; + } + if (!transportNameNode.isTextual()) { + warnAt(outerJson, connectorPath + "/transport/name", outerStart, reporter, + "Invalid connector", "\"transport.name\" must be a string"); + return; + } + JsonNode transportConfig = transport.get("config"); + if (transportConfig == null) + return; + if (!transportConfig.isObject()) { + warnAt(outerJson, connectorPath + "/transport/config", outerStart, reporter, + "Invalid connector", "\"transport.config\" must be a JSON object"); + return; + } + String configPointer = connectorPath + "/transport/config"; + String transportName = transportNameNode.asText(); + if (isTable) { + switch (transportName) { + case "delta_table_input": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, DeltaTableReaderConfig.class, reporter); + break; + case "iceberg_input": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, IcebergReaderConfig.class, reporter); + break; + case "kafka_input": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, KafkaInputConfig.class, reporter); + break; + case "postgres_input": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, PostgresReaderConfig.class, reporter); + break; + case "postgres_cdc_input": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, PostgresCdcReaderConfig.class, reporter); + break; + case "pub_sub_input": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, PubSubInputConfig.class, reporter); + break; + case "s3_input": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, S3InputConfig.class, reporter); + break; + case "clock": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, ClockConfig.class, reporter); + break; + case "file_input": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, FileInputConfig.class, reporter); + break; + case "url_input": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, UrlInputConfig.class, reporter); + break; + case "nats_input": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, NatsInputConfig.class, reporter); + break; + case "datagen": + case "nexmark": + case "empty": + case "adhoc": + break; + default: + warnAt(outerJson, connectorPath + "/transport/name", outerStart, reporter, + "Unknown format", "\"transport.name\" " + Utilities.doubleQuote(transportName, true) + " is not known"); + break; + } + } else { + switch (transportName) { + case "delta_table_output": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, DeltaTableWriterConfig.class, reporter); + break; + case "http_output": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, HttpOutputConfig.class, reporter); + break; + case "kafka_output": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, KafkaOutputConfig.class, reporter); + break; + case "postgres_output": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, PostgresWriterConfig.class, reporter); + break; + case "dynamodb_output": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, DynamoDBWriterConfig.class, reporter); + break; + case "file_output": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, FileOutputConfig.class, reporter); + break; + case "redis_output": + validateConfig(transportConfig, outerJson, configPointer, + outerStart, RedisOutputConfig.class, reporter); + break; + case "null": + break; + default: + warnAt(outerJson, connectorPath + "/transport/name", outerStart, reporter, + "Unknown format", "\"transport.name\" " + Utilities.doubleQuote(transportName, true) + " is not known"); + break; + } + } + } + + /** + * Deserializes a JSON config node into a typed POJO, reporting type errors + * and unknown fields as warnings via the error reporter. + * + * @param configNode The already-parsed JSON node to deserialize. + * @param outerJson The raw JSON string containing {@code configNode}. + * @param configPointer JSON Pointer path to {@code configNode} within {@code outerJson} + * (e.g. {@code "/0/format/config"}). + * @param outerStart Absolute source position of {@code outerJson}'s first character, + * used to convert within-JSON positions to source positions. + * @param clazz Target class to deserialize into. + * @param reporter Error reporter. + * @return Deserialized object, or {@code null} if validation errors were found. + */ + @Nullable + public static T validateConfig( + JsonNode configNode, + String outerJson, + String configPointer, + SourcePosition outerStart, + Class clazz, + IErrorReporter reporter) { + try { + T result = Utilities.deterministicObjectMapper() + .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, true) + .treeToValue(configNode, clazz); + var configReporter = new ConfigReporter(reporter, outerJson, configPointer, outerStart); + result.validate(configReporter); + return result; + } catch (JsonMappingException e) { + String fullPointer = configPointer + pathToPointer(e.getPath()); + + // Always point at the field key so the user sees exactly what to fix. + SourcePosition pos = Utilities.jsonPointerLocation(outerJson, fullPointer, false); + if (pos == null) + pos = Utilities.jsonPointerLocation(outerJson, configPointer, true); + SourcePositionRange range = pos != null + ? pos.relativeTo(outerStart).asRange() + : SourcePositionRange.INVALID; + + String message; + if (e instanceof UnrecognizedPropertyException upe) { + message = "unknown field " + Utilities.doubleQuote(upe.getPropertyName(), false); + } else if (e instanceof InvalidFormatException ife + && ife.getTargetType() != null && ife.getTargetType().isEnum()) { + String fieldName = pathToDotted(e.getPath()); + String validValues = enumValidValues(ife.getTargetType()); + message = (fieldName.isEmpty() ? "" : "field " + Utilities.doubleQuote(fieldName, false) + ": ") + + "invalid value " + Utilities.doubleQuote(String.valueOf(ife.getValue()), false) + + "; valid values are: " + validValues; + } else { + String field = pathToDotted(e.getPath()); + message = (field.isEmpty() ? "" : "field " + Utilities.doubleQuote(field, false) + ": ") + + e.getOriginalMessage(); + } + reporter.reportWarning(range, "Invalid configuration", message); + return null; + } catch (JsonProcessingException e) { + SourcePosition pos = Utilities.jsonPointerLocation(outerJson, configPointer, true); + SourcePositionRange range = pos != null + ? pos.relativeTo(outerStart).asRange() + : SourcePositionRange.INVALID; + reporter.reportWarning(range, "Invalid configuration", e.getOriginalMessage()); + return null; + } + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/IValidateConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/IValidateConfig.java new file mode 100644 index 00000000000..47419fd2efa --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/IValidateConfig.java @@ -0,0 +1,29 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors; + +import org.dbsp.util.Utilities; + +/** + * Validation interface for connector config POJOs. The reporter carries the + * JSON/source-position start of the validated document + * so that errors can point at the exact field that caused the problem. + */ +public interface IValidateConfig { + /** + * Validate the object after Jackson deserialization. + * + * @param reporter Field-aware error reporter; use {@link ConfigReporter#warnPath} to emit + * warnings with accurate source locations. + * @return {@code true} if the config is valid, {@code false} if at least one warning + * was emitted. + */ + boolean validate(ConfigReporter reporter); + + default boolean checkNonEmpty(ConfigReporter reporter, String property, String propertyName) { + if (property.isBlank()) { + reporter.warnPath(propertyName, "Invalid configuration", + "required field " + Utilities.doubleQuote(propertyName, false) + " is missing or empty"); + return false; + } + return true; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/AvroEncoderConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/AvroEncoderConfig.java new file mode 100644 index 00000000000..eae63b4ea77 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/AvroEncoderConfig.java @@ -0,0 +1,95 @@ +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; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** Configuration for the Avro output format. + * Registry fields are inlined directly. */ +@SuppressWarnings("unused") +public class AvroEncoderConfig implements IValidateConfig { + @JsonProperty("update_format") + public AvroUpdateFormat updateFormat = AvroUpdateFormat.Raw; + + @Nullable + @JsonProperty("key_mode") + public AvroEncoderKeyMode keyMode = null; + + @Nullable + @JsonProperty("schema") + public String schema = null; + + /** Only valid with {@code update_format = "raw"}. */ + @Nullable + @JsonProperty("cdc_field") + public String cdcField = null; + + @Nullable + @JsonProperty("namespace") + public String namespace = null; + + @Nullable + @JsonProperty("subject_name_strategy") + public SubjectNameStrategy subjectNameStrategy = null; + + @JsonProperty("skip_schema_id") + public boolean skipSchemaId = false; + + @JsonProperty("threads") + public int threads = 4; + + // ---- Schema registry fields (from AvroSchemaRegistryConfig) ---- + + @JsonProperty("registry_urls") + public List registryUrls = new ArrayList<>(); + + @JsonProperty("registry_headers") + public Map registryHeaders = new HashMap<>(); + + @Nullable + @JsonProperty("registry_proxy") + public String registryProxy = null; + + @Nullable + @JsonProperty("registry_timeout_secs") + public Long registryTimeoutSecs = null; + + /** Mutually exclusive with {@code registry_authorization_token}. */ + @Nullable + @JsonProperty("registry_username") + public String registryUsername = null; + + @Nullable + @JsonProperty("registry_password") + public String registryPassword = null; + + /** Mutually exclusive with {@code registry_username} / {@code registry_password}. */ + @Nullable + @JsonProperty("registry_authorization_token") + public String registryAuthorizationToken = null; + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + if (cdcField != null && updateFormat != AvroUpdateFormat.Raw) { + reporter.warnPath("cdc_field", "Invalid configuration", + "\"cdc_field\" is only valid with \"update_format\": \"raw\""); + ok = false; + } + if (registryAuthorizationToken != null + && (registryUsername != null || registryPassword != null)) { + reporter.warnPath("registry_authorization_token", "Invalid configuration", + "\"registry_authorization_token\" is mutually exclusive with " + + "\"registry_username\" and \"registry_password\""); + ok = false; + } + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/AvroEncoderKeyMode.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/AvroEncoderKeyMode.java new file mode 100644 index 00000000000..5cbe09e8cfe --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/AvroEncoderKeyMode.java @@ -0,0 +1,7 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum AvroEncoderKeyMode { + @JsonProperty("none") None, + @JsonProperty("key_fields") KeyFields, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/AvroParserConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/AvroParserConfig.java new file mode 100644 index 00000000000..7330abc62f5 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/AvroParserConfig.java @@ -0,0 +1,76 @@ +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; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** Configuration for the Avro input format. + * Registry fields are inlined directly. */ +@SuppressWarnings("unused") +public class AvroParserConfig implements IValidateConfig { + @JsonProperty("update_format") + public AvroUpdateFormat updateFormat = AvroUpdateFormat.Raw; + + /** Explicit Avro schema (JSON string). Mutually exclusive with {@code registry_urls}. */ + @Nullable + @JsonProperty("schema") + public String schema = null; + + @JsonProperty("skip_schema_id") + public boolean skipSchemaId = false; + + // ---- Schema registry fields (from AvroSchemaRegistryConfig) ---- + + @JsonProperty("registry_urls") + public List registryUrls = new ArrayList<>(); + + @JsonProperty("registry_headers") + public Map registryHeaders = new HashMap<>(); + + @Nullable + @JsonProperty("registry_proxy") + public String registryProxy = null; + + @Nullable + @JsonProperty("registry_timeout_secs") + public Long registryTimeoutSecs = null; + + /** Mutually exclusive with {@code registry_authorization_token}. */ + @Nullable + @JsonProperty("registry_username") + public String registryUsername = null; + + @Nullable + @JsonProperty("registry_password") + public String registryPassword = null; + + /** Mutually exclusive with {@code registry_username} / {@code registry_password}. */ + @Nullable + @JsonProperty("registry_authorization_token") + public String registryAuthorizationToken = null; + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + if (schema != null && !registryUrls.isEmpty()) { + reporter.warnPath("schema", "Invalid configuration", + "\"schema\" and \"registry_urls\" are mutually exclusive"); + ok = false; + } + if (registryAuthorizationToken != null + && (registryUsername != null || registryPassword != null)) { + reporter.warnPath("registry_authorization_token", "Invalid configuration", + "\"registry_authorization_token\" is mutually exclusive with " + + "\"registry_username\" and \"registry_password\""); + ok = false; + } + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/AvroUpdateFormat.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/AvroUpdateFormat.java new file mode 100644 index 00000000000..ac870483ad1 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/AvroUpdateFormat.java @@ -0,0 +1,8 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum AvroUpdateFormat { + @JsonProperty("raw") Raw, + @JsonProperty("debezium") Debezium, + @JsonProperty("confluent_jdbc") ConfluentJdbc, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/ClockConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/ClockConfig.java new file mode 100644 index 00000000000..0db91d4b892 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/ClockConfig.java @@ -0,0 +1,33 @@ +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; + +/** Configuration for the clock input connector. */ +@SuppressWarnings("unused") +public class ClockConfig implements IValidateConfig { + @JsonProperty("clock_resolution_usecs") + public long clockResolutionUsecs = 0; + + @Nullable + @JsonProperty("now_offset_ms") + public Long nowOffsetMs = null; + + @JsonProperty("http_driven") + public boolean httpDriven = false; + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + if (clockResolutionUsecs <= 0) { + reporter.warnPath("clock_resolution_usecs", "Invalid configuration", + "\"clock_resolution_usecs\" must be greater than 0"); + ok = false; + } + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/CsvEncoderConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/CsvEncoderConfig.java new file mode 100644 index 00000000000..a18162a8c6b --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/CsvEncoderConfig.java @@ -0,0 +1,30 @@ +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 org.dbsp.util.Utilities; + +/** Configuration for the CSV output format of a connector */ +@SuppressWarnings("unused") +public class CsvEncoderConfig implements IValidateConfig { + /** Field delimiter. Must be an ASCII character. */ + @JsonProperty("delimiter") + public char delimiter = ','; + + /** Number of records to buffer before flushing output. */ + @JsonProperty("buffer_size_records") + public long bufferSizeRecords = 10_000; + + @Override + public boolean validate(ConfigReporter reporter) { + if (delimiter > 127) { + reporter.warnPath("delimiter", "Invalid configuration", + "field \"delimiter\" must be an ASCII character; got " + + Utilities.singleQuote(String.valueOf(delimiter))); + return false; + } + return true; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/CsvParserConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/CsvParserConfig.java new file mode 100644 index 00000000000..0a42215a4f6 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/CsvParserConfig.java @@ -0,0 +1,81 @@ +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 org.dbsp.util.Utilities; + +import javax.annotation.Nullable; + +/** Configuration for the CSV input connector format. */ +@SuppressWarnings("unused") +public class CsvParserConfig implements IValidateConfig { + /** Field delimiter. Must be an ASCII character. */ + @JsonProperty("delimiter") + public char delimiter = ','; + + /** Whether the input begins with a header line, which is skipped. */ + @JsonProperty("headers") + public boolean headers = false; + + /** Quote character. Must be an ASCII character. */ + @JsonProperty("quote") + public char quote = '"'; + + /** Escape character for quoted fields. + * When {@code null} (the default), doubled quotes are used instead. */ + @Nullable + @JsonProperty("escape") + public Character escape = null; + + /** Enable double-quote escaping inside quoted fields. */ + @JsonProperty("double_quote") + public boolean doubleQuote = true; + + /** Enable quoting. When {@code false}, every newline terminates a record. */ + @JsonProperty("quoting") + public boolean quoting = true; + + /** Lines starting with this character are treated as comments and skipped. + * {@code null} disables comment handling. */ + @Nullable + @JsonProperty("comment") + public Character comment = null; + + /** Allow records with a variable number of fields. */ + @JsonProperty("flexible") + public boolean flexible = true; + + /** Whitespace trimming policy. */ + @JsonProperty("trim") + public CsvTrim trim = CsvTrim.None; + + /** Additional constraints: all single-character fields must be ASCII + * (the underlying CSV library uses single-byte characters), and {@code delimiter} and + * {@code quote} must be distinct. */ + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + for (var entry : new Object[][]{ + {"delimiter", delimiter}, + {"quote", quote}, + {"escape", escape}, + {"comment", comment}}) { + String name = (String) entry[0]; + Character c = (Character) entry[1]; + if (c != null && c > 127) { + reporter.warnPath(name, "Invalid configuration", + "field " + Utilities.doubleQuote(name, false) + + " must be an ASCII character; got " + Utilities.singleQuote(c.toString())); + ok = false; + } + } + if (ok && delimiter == quote) { + reporter.warnPath("delimiter", "Invalid configuration", + "fields \"delimiter\" and \"quote\" must be different characters"); + ok = false; + } + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/CsvTrim.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/CsvTrim.java new file mode 100644 index 00000000000..e8dcb3bea41 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/CsvTrim.java @@ -0,0 +1,10 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** Whitespace trimming policy applied to CSV fields or header names. */ +public enum CsvTrim { + @JsonProperty("none") None, + @JsonProperty("headers") Headers, + @JsonProperty("fields") Fields, + @JsonProperty("all") All, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableIngestMode.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableIngestMode.java new file mode 100644 index 00000000000..7401b17f887 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableIngestMode.java @@ -0,0 +1,9 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum DeltaTableIngestMode { + @JsonProperty("snapshot") Snapshot, + @JsonProperty("follow") Follow, + @JsonProperty("snapshot_and_follow") SnapshotAndFollow, + @JsonProperty("cdc") Cdc, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableReaderConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableReaderConfig.java new file mode 100644 index 00000000000..e0e4f543a7f --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableReaderConfig.java @@ -0,0 +1,116 @@ +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.JsonAlias; +import com.fasterxml.jackson.annotation.JsonAnySetter; +import com.fasterxml.jackson.annotation.JsonProperty; + +import javax.annotation.Nullable; + +/** Configuration for the Delta Table input transport. */ +@SuppressWarnings("unused") +public class DeltaTableReaderConfig implements IValidateConfig { + @JsonProperty("uri") + public String uri = ""; + + /** Required. */ + @Nullable + @JsonProperty("mode") + public DeltaTableIngestMode mode = null; + + @JsonProperty("transaction_mode") + public DeltaTableTransactionMode transactionMode = DeltaTableTransactionMode.None; + + @Nullable + @JsonProperty("timestamp_column") + public String timestampColumn = null; + + @Nullable + @JsonProperty("filter") + public String filter = null; + + @JsonProperty("skip_unused_columns") + public boolean skipUnusedColumns = false; + + /** Only valid when {@code mode} is {@code snapshot} or {@code snapshot_and_follow}. */ + @Nullable + @JsonProperty("snapshot_filter") + public String snapshotFilter = null; + + @Nullable + @JsonAlias("start_version") + @JsonProperty("version") + public Long version = null; + + @Nullable + @JsonAlias("start_datetime") + @JsonProperty("datetime") + public String datetime = null; + + @Nullable + @JsonProperty("end_version") + public Long endVersion = null; + + /** Only valid in {@code cdc} mode. */ + @Nullable + @JsonProperty("cdc_delete_filter") + public String cdcDeleteFilter = null; + + /** Only valid in {@code cdc} mode. */ + @Nullable + @JsonProperty("cdc_order_by") + public String cdcOrderBy = null; + + @JsonProperty("num_parsers") + public int numParsers = 4; + + @Nullable + @JsonProperty("max_concurrent_readers") + public Long maxConcurrentReaders = null; + + @JsonProperty("verbose") + public int verbose = 0; + + @Nullable + @JsonProperty("max_retries") + public Long maxRetries = null; + + /** Absorbs flattened {@code object_store_config} keys so they are not rejected as unknown. */ + @JsonAnySetter + public void setObjectStoreOption(String key, Object value) {} + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + if (mode == null) { + reporter.warn("Invalid configuration", "required field \"mode\" is missing"); + return false; + } + if (version != null && datetime != null) { + reporter.warnPath("version", "Invalid configuration", + "\"version\" and \"datetime\" are mutually exclusive"); + ok = false; + } + if (snapshotFilter != null + && mode != DeltaTableIngestMode.Snapshot + && mode != DeltaTableIngestMode.SnapshotAndFollow) { + reporter.warnPath("snapshot_filter", "Invalid configuration", + "\"snapshot_filter\" is only valid with \"mode\": \"snapshot\" or \"snapshot_and_follow\""); + ok = false; + } + if (cdcDeleteFilter != null && mode != DeltaTableIngestMode.Cdc) { + reporter.warnPath("cdc_delete_filter", "Invalid configuration", + "\"cdc_delete_filter\" is only valid with \"mode\": \"cdc\""); + ok = false; + } + if (cdcOrderBy != null && mode != DeltaTableIngestMode.Cdc) { + reporter.warnPath("cdc_order_by", "Invalid configuration", + "\"cdc_order_by\" is only valid with \"mode\": \"cdc\""); + ok = false; + } + ok = ok && this.checkNonEmpty(reporter, this.uri, "uri"); + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableTransactionMode.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableTransactionMode.java new file mode 100644 index 00000000000..de0e426666b --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableTransactionMode.java @@ -0,0 +1,9 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum DeltaTableTransactionMode { + @JsonProperty("none") None, + @JsonProperty("snapshot") Snapshot, + @JsonProperty("always") Always, + @JsonProperty("catchup") Catchup, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableWriteMode.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableWriteMode.java new file mode 100644 index 00000000000..d3284476f16 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableWriteMode.java @@ -0,0 +1,8 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum DeltaTableWriteMode { + @JsonProperty("append") Append, + @JsonProperty("truncate") Truncate, + @JsonProperty("error_if_exists") ErrorIfExists, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableWriterConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableWriterConfig.java new file mode 100644 index 00000000000..ceafc7ea358 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DeltaTableWriterConfig.java @@ -0,0 +1,95 @@ +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.JsonAnySetter; +import com.fasterxml.jackson.annotation.JsonProperty; + +import javax.annotation.Nullable; +import java.util.Set; + +/** Configuration for the Delta Table output transport. */ +@SuppressWarnings("unused") +public class DeltaTableWriterConfig implements IValidateConfig { + private static final Set VALID_INTERVAL_UNITS = Set.of( + "nanosecond", "nanoseconds", "microsecond", "microseconds", + "millisecond", "milliseconds", "second", "seconds", + "minute", "minutes", "hour", "hours", "day", "days", "week", "weeks"); + + @JsonProperty("uri") + public String uri = ""; + + @JsonProperty("mode") + public DeltaTableWriteMode mode = DeltaTableWriteMode.Append; + + @Nullable + @JsonProperty("checkpoint_interval") + public Long checkpointInterval = null; + + /** Must follow Delta Lake interval syntax: {@code "interval "}. */ + @Nullable + @JsonProperty("log_retention_duration") + public String logRetentionDuration = null; + + @Nullable + @JsonProperty("enable_expired_log_cleanup") + public Boolean enableExpiredLogCleanup = null; + + @Nullable + @JsonProperty("max_retries") + public Long maxRetries = null; + + /** Must be {@code > 0} when set. */ + @Nullable + @JsonProperty("threads") + public Long threads = null; + + /** Absorbs flattened {@code object_store_config} keys so they are not rejected as unknown. */ + @JsonAnySetter + public void setObjectStoreOption(String key, Object value) {} + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + if (threads != null && threads == 0) { + reporter.warnPath("threads", "Invalid configuration", + "\"threads\" must be greater than 0"); + ok = false; + } + if (logRetentionDuration != null) { + String error = validateDeltaInterval(logRetentionDuration); + if (error != null) { + reporter.warnPath("log_retention_duration", "Invalid configuration", + "invalid \"log_retention_duration\" value \"" + + logRetentionDuration + "\": " + error); + ok = false; + } + } + return ok; + } + + /** Validates a Delta Lake interval string (e.g. {@code "interval 30 days"}). + * @return an error message, or {@code null} if the value is valid. */ + @Nullable + static String validateDeltaInterval(String value) { + String[] tokens = value.strip().split("\\s+"); + if (tokens.length != 3) + return "expected format \"interval \" (e.g. \"interval 30 days\")"; + if (!tokens[0].equals("interval")) + return "expected the value to start with lowercase \"interval\""; + long number; + try { + number = Long.parseLong(tokens[1]); + } catch (NumberFormatException e) { + return "cannot parse '" + tokens[1] + "' as integer: " + e.getMessage(); + } + if (number < 0) + return "interval cannot be negative"; + if (!VALID_INTERVAL_UNITS.contains(tokens[2])) + return "unknown unit '" + tokens[2] + + "'; expected one of nanosecond[s], microsecond[s], millisecond[s]," + + " second[s], minute[s], hour[s], day[s], week[s]"; + return null; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DynamoDBWriteMode.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DynamoDBWriteMode.java new file mode 100644 index 00000000000..5ab5effdc1b --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DynamoDBWriteMode.java @@ -0,0 +1,8 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; + +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum DynamoDBWriteMode { + @JsonProperty("batch") Batch, + @JsonProperty("transactional") Transactional, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DynamoDBWriterConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DynamoDBWriterConfig.java new file mode 100644 index 00000000000..afa61a1125c --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/DynamoDBWriterConfig.java @@ -0,0 +1,110 @@ +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; + +/** Configuration for the DynamoDB output connector. */ +@SuppressWarnings("unused") +public class DynamoDBWriterConfig implements IValidateConfig { + private static final int BATCH_MAX = 25; + private static final int TRANSACTIONAL_MAX = 100; + + @JsonProperty("table") + public String table = ""; + + @JsonProperty("region") + public String region = ""; + + @Nullable + @JsonProperty("endpoint_url") + public String endpointUrl = null; + + @Nullable + @JsonProperty("aws_access_key_id") + public String awsAccessKeyId = null; + + @Nullable + @JsonProperty("aws_secret_access_key") + public String awsSecretAccessKey = null; + + @Nullable + @JsonProperty("batch_size") + public Integer batchSize = null; + + @JsonProperty("write_mode") + public DynamoDBWriteMode writeMode = DynamoDBWriteMode.Batch; + + @JsonProperty("max_buffer_size_bytes") + public long maxBufferSizeBytes = 1024 * 1024; + + @JsonProperty("max_concurrent_requests") + public int maxConcurrentRequests = 64; + + @JsonProperty("threads") + public int threads = 1; + + @Nullable + @JsonProperty("max_retries") + public Integer maxRetries = 10; + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + if (table.isBlank()) { + reporter.warnPath("table", "Invalid configuration", + "\"table\" cannot be empty"); + ok = false; + } else if (table.length() < 3 || table.length() > 255) { + reporter.warnPath("table", "Invalid configuration", + "table name \"" + table + "\" is " + table.length() + + " character(s); DynamoDB requires 3–255 characters"); + ok = false; + } else if (!table.chars().allMatch( + c -> (Character.isLetterOrDigit(c) && c < 0x80) || c == '_' || c == '-' || c == '.')) { + reporter.warnPath("table", "Invalid configuration", + "table name \"" + table + "\" contains invalid characters;" + + " DynamoDB requires [a-zA-Z0-9_.-]"); + ok = false; + } + if (region.isBlank()) { + reporter.warnPath("region", "Invalid configuration", + "\"region\" cannot be empty"); + ok = false; + } + if (batchSize != null) { + int maxBatchSize = writeMode == DynamoDBWriteMode.Batch ? BATCH_MAX : TRANSACTIONAL_MAX; + if (batchSize == 0 || batchSize > maxBatchSize) { + reporter.warnPath("batch_size", "Invalid configuration", + "\"batch_size\" must be between 1 and " + maxBatchSize + + " for \"" + (writeMode == DynamoDBWriteMode.Batch ? "batch" : "transactional") + + "\" write mode"); + ok = false; + } + } + if (threads == 0) { + reporter.warnPath("threads", "Invalid configuration", + "\"threads\" must be at least 1"); + ok = false; + } + if (maxBufferSizeBytes == 0) { + reporter.warnPath("max_buffer_size_bytes", "Invalid configuration", + "\"max_buffer_size_bytes\" must be greater than 0"); + ok = false; + } + if (maxConcurrentRequests == 0) { + reporter.warnPath("max_concurrent_requests", "Invalid configuration", + "\"max_concurrent_requests\" must be greater than 0"); + ok = false; + } + if ((awsAccessKeyId == null) != (awsSecretAccessKey == null)) { + reporter.warn("Invalid configuration", + "\"aws_access_key_id\" and \"aws_secret_access_key\" must be specified together"); + ok = false; + } + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/FileInputConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/FileInputConfig.java new file mode 100644 index 00000000000..ac5483ff59b --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/FileInputConfig.java @@ -0,0 +1,27 @@ +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; + +/** Configuration for reading data from a file. */ +@SuppressWarnings("unused") +public class FileInputConfig implements IValidateConfig { + @JsonProperty("path") + public String path = ""; + + @Nullable + @JsonProperty("buffer_size_bytes") + public Long bufferSizeBytes = null; + + @JsonProperty("follow") + public boolean follow = false; + + @Override + public boolean validate(ConfigReporter reporter) { + return true; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/FileOutputConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/FileOutputConfig.java new file mode 100644 index 00000000000..16080dde3f0 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/FileOutputConfig.java @@ -0,0 +1,18 @@ +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; + +/** Configuration for writing data to a file. */ +@SuppressWarnings("unused") +public class FileOutputConfig implements IValidateConfig { + @JsonProperty("path") + public String path = ""; + + @Override + public boolean validate(ConfigReporter reporter) { + return true; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/HttpOutputConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/HttpOutputConfig.java new file mode 100644 index 00000000000..1a536e61da2 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/HttpOutputConfig.java @@ -0,0 +1,20 @@ +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; + +/** Configuration for data output via HTTP. */ +@SuppressWarnings("unused") +public class HttpOutputConfig implements IValidateConfig { + /** When {@code true}, block the pipeline if the HTTP client cannot keep up. + * When {@code false} (the default), drop chunks instead of blocking. */ + @JsonProperty("backpressure") + public boolean backpressure = false; + + @Override + public boolean validate(ConfigReporter reporter) { + return true; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergCatalogType.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergCatalogType.java new file mode 100644 index 00000000000..96cd2fe4b01 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergCatalogType.java @@ -0,0 +1,7 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum IcebergCatalogType { + @JsonProperty("rest") Rest, + @JsonProperty("glue") Glue, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergIngestMode.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergIngestMode.java new file mode 100644 index 00000000000..476a137121a --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergIngestMode.java @@ -0,0 +1,8 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum IcebergIngestMode { + @JsonProperty("snapshot") Snapshot, + @JsonProperty("follow") Follow, + @JsonProperty("snapshot_and_follow") SnapshotAndFollow, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergReaderConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergReaderConfig.java new file mode 100644 index 00000000000..24ecc59e328 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergReaderConfig.java @@ -0,0 +1,160 @@ +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.JsonAnySetter; +import com.fasterxml.jackson.annotation.JsonProperty; + +import javax.annotation.Nullable; + +/** Configuration for reading data from an Iceberg table. */ +@SuppressWarnings("unused") +public class IcebergReaderConfig implements IValidateConfig { + /** Required. */ + @Nullable + @JsonProperty("mode") + public IcebergIngestMode mode = null; + + @Nullable + @JsonProperty("timestamp_column") + public String timestampColumn = null; + + @Nullable + @JsonProperty("snapshot_filter") + public String snapshotFilter = null; + + @Nullable + @JsonProperty("snapshot_id") + public Long snapshotId = null; + + @Nullable + @JsonProperty("datetime") + public String datetime = null; + + @Nullable + @JsonProperty("metadata_location") + public String metadataLocation = null; + + @Nullable + @JsonProperty("table_name") + public String tableName = null; + + @Nullable + @JsonProperty("catalog_type") + public IcebergCatalogType catalogType = null; + + // Glue catalog fields (inlined from GlueCatalogConfig) + @Nullable @JsonProperty("glue.warehouse") public String glueWarehouse = null; + @Nullable @JsonProperty("glue.endpoint") public String glueEndpoint = null; + @Nullable @JsonProperty("glue.access-key-id") public String glueAccessKeyId = null; + @Nullable @JsonProperty("glue.secret-access-key") public String glueSecretAccessKey = null; + @Nullable @JsonProperty("glue.profile-name") public String glueProfileName = null; + @Nullable @JsonProperty("glue.region") public String glueRegion = null; + @Nullable @JsonProperty("glue.session-token") public String glueSessionToken = null; + @Nullable @JsonProperty("glue.id") public String glueId = null; + + // REST catalog fields (inlined from RestCatalogConfig) + @Nullable @JsonProperty("rest.uri") public String restUri = null; + @Nullable @JsonProperty("rest.warehouse") public String restWarehouse = null; + @Nullable @JsonProperty("rest.oauth2-server-uri") public String restOauth2ServerUri = null; + @Nullable @JsonProperty("rest.credential") public String restCredential = null; + @Nullable @JsonProperty("rest.token") public String restToken = null; + @Nullable @JsonProperty("rest.scope") public String restScope = null; + @Nullable @JsonProperty("rest.prefix") public String restPrefix = null; + @Nullable @JsonProperty("rest.headers") public Object restHeaders = null; + @Nullable @JsonProperty("rest.audience") public String restAudience = null; + @Nullable @JsonProperty("rest.resource") public String restResource = null; + + /** Absorbs flattened {@code fileio_config} keys so they are not rejected as unknown. */ + @JsonAnySetter + public void setFileioOption(String key, Object value) {} + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + if (mode == null) { + reporter.warn("Invalid configuration", "required field \"mode\" is missing"); + return false; + } + if (snapshotId != null && datetime != null) { + reporter.warnPath("snapshot_id", "Invalid configuration", + "\"snapshot_id\" and \"datetime\" are mutually exclusive"); + ok = false; + } + if (catalogType == null && metadataLocation == null) { + reporter.warn("Invalid configuration", + "missing metadata location: specify an Iceberg catalog via \"catalog_type\"" + + " or provide a table metadata location via \"metadata_location\""); + ok = false; + } else if (catalogType != null && metadataLocation != null) { + reporter.warnPath("metadata_location", "Invalid configuration", + "\"metadata_location\" is not supported when \"catalog_type\" is set"); + ok = false; + } + if (catalogType == null && tableName != null) { + reporter.warnPath("table_name", "Invalid configuration", + "\"table_name\" is only valid when \"catalog_type\" is set"); + ok = false; + } else if (catalogType != null && tableName == null) { + reporter.warn("Invalid configuration", + "\"table_name\" must be specified when \"catalog_type\" is set"); + ok = false; + } + if (catalogType == IcebergCatalogType.Glue) { + if (glueWarehouse == null) { + reporter.warn("Invalid configuration", + "missing Iceberg warehouse location—set \"glue.warehouse\" when using" + + " \"catalog_type\": \"glue\""); + ok = false; + } + } else { + ok &= checkGluePropAbsent(reporter, glueWarehouse, "warehouse"); + ok &= checkGluePropAbsent(reporter, glueEndpoint, "endpoint"); + ok &= checkGluePropAbsent(reporter, glueAccessKeyId, "access-key-id"); + ok &= checkGluePropAbsent(reporter, glueSecretAccessKey, "secret-access-key"); + ok &= checkGluePropAbsent(reporter, glueProfileName, "profile-name"); + ok &= checkGluePropAbsent(reporter, glueRegion, "region"); + ok &= checkGluePropAbsent(reporter, glueSessionToken, "session-token"); + ok &= checkGluePropAbsent(reporter, glueId, "id"); + } + if (catalogType == IcebergCatalogType.Rest) { + if (restUri == null) { + reporter.warn("Invalid configuration", + "missing Iceberg REST catalog URI—set \"rest.uri\" when using" + + " \"catalog_type\": \"rest\""); + ok = false; + } + } else { + ok &= checkRestPropAbsent(reporter, restUri, "uri"); + ok &= checkRestPropAbsent(reporter, restWarehouse, "warehouse"); + ok &= checkRestPropAbsent(reporter, restOauth2ServerUri, "oauth2-server-uri"); + ok &= checkRestPropAbsent(reporter, restCredential, "credential"); + ok &= checkRestPropAbsent(reporter, restToken, "token"); + ok &= checkRestPropAbsent(reporter, restScope, "scope"); + ok &= checkRestPropAbsent(reporter, restPrefix, "prefix"); + ok &= checkRestPropAbsent(reporter, restHeaders, "headers"); + ok &= checkRestPropAbsent(reporter, restAudience, "audience"); + ok &= checkRestPropAbsent(reporter, restResource, "resource"); + } + return ok; + } + + private boolean checkGluePropAbsent(ConfigReporter reporter, Object value, String name) { + if (value != null) { + reporter.warnPath("glue." + name, "Invalid configuration", + "\"glue." + name + "\" is only valid when \"catalog_type\": \"glue\""); + return false; + } + return true; + } + + private boolean checkRestPropAbsent(ConfigReporter reporter, Object value, String name) { + if (value != null) { + reporter.warnPath("rest." + name, "Invalid configuration", + "\"rest." + name + "\" is only valid when \"catalog_type\": \"rest\""); + return false; + } + return true; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonEncoderConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonEncoderConfig.java new file mode 100644 index 00000000000..0a8b48e9371 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonEncoderConfig.java @@ -0,0 +1,42 @@ +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; +import java.util.List; + +/** Configuration for the JSON output format. */ +@SuppressWarnings("unused") +public class JsonEncoderConfig implements IValidateConfig { + @JsonProperty("update_format") + public JsonUpdateFormat updateFormat = JsonUpdateFormat.InsertDelete; + + @Nullable + @JsonProperty("json_flavor") + public JsonFlavor jsonFlavor = null; + + @JsonProperty("buffer_size_records") + public long bufferSizeRecords = 10_000; + + @JsonProperty("array") + public boolean array = false; + + /** When set, only these columns appear in the Debezium message key. + * Valid only with {@code update_format = "debezium"}. */ + @Nullable + @JsonProperty("key_fields") + public List keyFields = null; + + @Override + public boolean validate(ConfigReporter reporter) { + if (keyFields != null && updateFormat != JsonUpdateFormat.Debezium) { + reporter.warnPath("key_fields", "Invalid configuration", + "\"key_fields\" is only valid with \"update_format\": \"debezium\""); + return false; + } + return true; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonFlavor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonFlavor.java new file mode 100644 index 00000000000..cbb6dc4cb35 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonFlavor.java @@ -0,0 +1,19 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** Mirrors the Rust {@code JsonFlavor} enum. + * The {@code ParquetConverter} variant is omitted: its {@code #[serde(skip)]} attribute + * means serde will never serialize or deserialize it, so it cannot appear in JSON. */ +public enum JsonFlavor { + @JsonProperty("default") Default, + @JsonProperty("debezium_mysql") DebeziumMySql, + @JsonProperty("debezium_postgres") DebeziumPostgres, + @JsonProperty("snowflake") Snowflake, + @JsonProperty("kafka_connect_json_converter") KafkaConnectJsonConverter, + @JsonProperty("pandas") Pandas, + @JsonProperty("blockchain") Blockchain, + @JsonProperty("c_hex") CHex, + @JsonProperty("ClockInput") ClockInput, + @JsonProperty("datagen") Datagen, + @JsonProperty("postgres") Postgres, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonLines.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonLines.java new file mode 100644 index 00000000000..af5346c9236 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonLines.java @@ -0,0 +1,7 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum JsonLines { + @JsonProperty("multiple") Multiple, + @JsonProperty("single") Single, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonParserConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonParserConfig.java new file mode 100644 index 00000000000..998200d179a --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonParserConfig.java @@ -0,0 +1,29 @@ +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; + +/** Configuration for the JSON input format. */ +@SuppressWarnings("unused") +public class JsonParserConfig implements IValidateConfig { + @JsonProperty("update_format") + public JsonUpdateFormat updateFormat = JsonUpdateFormat.InsertDelete; + + @JsonProperty("json_flavor") + public JsonFlavor jsonFlavor = JsonFlavor.Default; + + /** When {@code true}, each input chunk is a JSON array of update objects. */ + @JsonProperty("array") + public boolean array = false; + + /** Whether JSON values may span multiple lines. */ + @JsonProperty("lines") + public JsonLines lines = JsonLines.Multiple; + + @Override + public boolean validate(ConfigReporter reporter) { + return true; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonUpdateFormat.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonUpdateFormat.java new file mode 100644 index 00000000000..e2cef062a94 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/JsonUpdateFormat.java @@ -0,0 +1,11 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum JsonUpdateFormat { + @JsonProperty("insert_delete") InsertDelete, + @JsonProperty("weighted") Weighted, + @JsonProperty("debezium") Debezium, + @JsonProperty("snowflake") Snowflake, + @JsonProperty("raw") Raw, + @JsonProperty("redis") Redis, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/KafkaInputConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/KafkaInputConfig.java new file mode 100644 index 00000000000..9815e76d05c --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/KafkaInputConfig.java @@ -0,0 +1,83 @@ +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.JsonAnySetter; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.JsonNode; + +import javax.annotation.Nullable; +import java.util.List; + +/** Configuration for reading data from a Kafka topic. */ +@SuppressWarnings("unused") +public class KafkaInputConfig implements IValidateConfig { + @JsonProperty("topic") + public String topic = ""; + + @Nullable + @JsonProperty("log_level") + public KafkaLogLevel logLevel = null; + + @JsonProperty("group_join_timeout_secs") + public int groupJoinTimeoutSecs = 10; + + @Nullable + @JsonProperty("poller_threads") + public Integer pollerThreads = null; + + /** Complex enum with data variants; validated structurally by the runtime. */ + @Nullable + @JsonProperty("start_from") + public JsonNode startFrom = null; + + @Nullable + @JsonProperty("region") + public String region = null; + + @Nullable + @JsonProperty("partitions") + public List partitions = null; + + @JsonProperty("resume_earliest_if_data_expires") + public boolean resumeEarliestIfDataExpires = false; + + @Nullable + @JsonProperty("include_headers") + public Boolean includeHeaders = null; + + @Nullable + @JsonProperty("include_timestamp") + public Boolean includeTimestamp = null; + + @Nullable + @JsonProperty("include_partition") + public Boolean includePartition = null; + + @Nullable + @JsonProperty("include_offset") + public Boolean includeOffset = null; + + @Nullable + @JsonProperty("include_topic") + public Boolean includeTopic = null; + + @JsonProperty("synchronize_partitions") + public boolean synchronizePartitions = false; + + /** Absorbs flattened {@code kafka_options} keys so they are not rejected as unknown. */ + @JsonAnySetter + public void setKafkaOption(String key, Object value) {} + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + if (topic.isBlank()) { + reporter.warn("Invalid configuration", + "required field \"topic\" is missing or empty"); + ok = false; + } + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/KafkaLogLevel.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/KafkaLogLevel.java new file mode 100644 index 00000000000..cde4be34186 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/KafkaLogLevel.java @@ -0,0 +1,13 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum KafkaLogLevel { + @JsonProperty("emerg") Emerg, + @JsonProperty("alert") Alert, + @JsonProperty("critical") Critical, + @JsonProperty("error") Error, + @JsonProperty("warning") Warning, + @JsonProperty("notice") Notice, + @JsonProperty("info") Info, + @JsonProperty("debug") Debug, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/KafkaOutputConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/KafkaOutputConfig.java new file mode 100644 index 00000000000..9c6223eb3f8 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/KafkaOutputConfig.java @@ -0,0 +1,56 @@ +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.JsonAnySetter; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.JsonNode; + +import javax.annotation.Nullable; +import java.util.List; + +/** Configuration for writing data to a Kafka topic. */ +@SuppressWarnings("unused") +public class KafkaOutputConfig implements IValidateConfig { + @JsonProperty("topic") + public String topic = ""; + + @Nullable + @JsonProperty("headers") + public List headers = null; + + @Nullable + @JsonProperty("log_level") + public KafkaLogLevel logLevel = null; + + @JsonProperty("initialization_timeout_secs") + public int initializationTimeoutSecs = 60; + + @Nullable + @JsonProperty("fault_tolerance") + public JsonNode faultTolerance = null; + + @Nullable + @JsonProperty("kafka_service") + public String kafkaService = null; + + @Nullable + @JsonProperty("region") + public String region = null; + + /** Absorbs flattened {@code kafka_options} keys so they are not rejected as unknown. */ + @JsonAnySetter + public void setKafkaOption(String key, Object value) {} + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + if (topic.isBlank()) { + reporter.warn("Invalid configuration", + "required field \"topic\" is missing or empty"); + ok = false; + } + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsAuthConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsAuthConfig.java new file mode 100644 index 00000000000..057bd687240 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsAuthConfig.java @@ -0,0 +1,37 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; + +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.JsonNode; + +import javax.annotation.Nullable; + +/** + * Authentication configuration for a NATS connection. + * + *

{@code credentials} maps to the Rust {@code Credentials} enum, which has + * data-carrying variants ({@code FromString(String)} and {@code FromFile(PathBuf)}) that + * serialize as {@code {"FromString": "..."}} or {@code {"FromFile": "..."}}. These are + * kept as {@link JsonNode} to avoid complex Jackson polymorphic deserialization. + */ +@SuppressWarnings("unused") +public class NatsAuthConfig { + @Nullable + @JsonProperty("credentials") + public JsonNode credentials = null; + + @Nullable + @JsonProperty("jwt") + public String jwt = null; + + @Nullable + @JsonProperty("nkey") + public String nkey = null; + + @Nullable + @JsonProperty("token") + public String token = null; + + @Nullable + @JsonProperty("user_and_password") + public NatsUserAndPasswordConfig userAndPassword = null; +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsConnectOptionsConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsConnectOptionsConfig.java new file mode 100644 index 00000000000..22c988d048b --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsConnectOptionsConfig.java @@ -0,0 +1,20 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; + +import com.fasterxml.jackson.annotation.JsonProperty; + +/** Options for connecting to a NATS server. */ +@SuppressWarnings("unused") +public class NatsConnectOptionsConfig { + /** NATS server URL (e.g., {@code "nats://localhost:4222"}). */ + @JsonProperty("server_url") + public String serverUrl = ""; + + @JsonProperty("auth") + public NatsAuthConfig auth = new NatsAuthConfig(); + + @JsonProperty("connection_timeout_secs") + public long connectionTimeoutSecs = 10; + + @JsonProperty("request_timeout_secs") + public long requestTimeoutSecs = 10; +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsConsumerConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsConsumerConfig.java new file mode 100644 index 00000000000..94227b482cb --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsConsumerConfig.java @@ -0,0 +1,57 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; + +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.JsonNode; + +import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * JetStream consumer configuration for a NATS input connector. + * Fields typed as {@link JsonNode} are accepted as-is and validated more thoroughly at runtime. + */ +@SuppressWarnings("unused") +public class NatsConsumerConfig { + @Nullable + @JsonProperty("name") + public String name = null; + + @Nullable + @JsonProperty("description") + public String description = null; + + @JsonProperty("filter_subjects") + public List filterSubjects = new ArrayList<>(); + + @JsonProperty("replay_policy") + public NatsReplayPolicy replayPolicy = NatsReplayPolicy.Instant; + + @JsonProperty("rate_limit") + public long rateLimit = 0; + + /** Required. */ + @Nullable + @JsonProperty("deliver_policy") + public JsonNode deliverPolicy = null; + + @JsonProperty("max_waiting") + public long maxWaiting = 0; + + @JsonProperty("metadata") + public Map metadata = new HashMap<>(); + + @Nullable + @JsonProperty("max_batch") + public Long maxBatch = null; + + @Nullable + @JsonProperty("max_bytes") + public Long maxBytes = null; + + @Nullable + @JsonProperty("max_expires") + public JsonNode maxExpires = null; +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsInputConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsInputConfig.java new file mode 100644 index 00000000000..0cee813f9ec --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsInputConfig.java @@ -0,0 +1,67 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; + +import com.fasterxml.jackson.annotation.JsonProperty; +import org.dbsp.sqlCompiler.compiler.frontend.connectors.ConfigReporter; +import org.dbsp.sqlCompiler.compiler.frontend.connectors.IValidateConfig; + +import javax.annotation.Nullable; + +/** Configuration for the NATS JetStream input connector. */ +@SuppressWarnings("unused") +public class NatsInputConfig implements IValidateConfig { + @Nullable + @JsonProperty("connection_config") + public NatsConnectOptionsConfig connectionConfig = null; + + @JsonProperty("stream_name") + public String streamName = ""; + + @JsonProperty("inactivity_timeout_secs") + public long inactivityTimeoutSecs = 60; + + @JsonProperty("retry_interval_secs") + public long retryIntervalSecs = 5; + + @Nullable + @JsonProperty("consumer_config") + public NatsConsumerConfig consumerConfig = null; + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + if (streamName.isBlank()) { + reporter.warnPath("stream_name", "Invalid configuration", + "required field \"stream_name\" is missing or empty"); + ok = false; + } + if (inactivityTimeoutSecs < 1) { + reporter.warnPath("inactivity_timeout_secs", "Invalid configuration", + "\"inactivity_timeout_secs\" must be at least 1"); + ok = false; + } + if (retryIntervalSecs < 1) { + reporter.warnPath("retry_interval_secs", "Invalid configuration", + "\"retry_interval_secs\" must be at least 1"); + ok = false; + } + if (connectionConfig == null) { + reporter.warn("Invalid configuration", + "required field \"connection_config\" is missing"); + ok = false; + } else if (connectionConfig.serverUrl.isBlank()) { + reporter.warnPath("connection_config/server_url", "Invalid configuration", + "required field \"connection_config.server_url\" is missing or empty"); + ok = false; + } + if (consumerConfig == null) { + reporter.warn("Invalid configuration", + "required field \"consumer_config\" is missing"); + ok = false; + } else if (consumerConfig.deliverPolicy == null || consumerConfig.deliverPolicy.isNull()) { + reporter.warnPath("consumer_config/deliver_policy", "Invalid configuration", + "required field \"consumer_config.deliver_policy\" is missing"); + ok = false; + } + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsReplayPolicy.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsReplayPolicy.java new file mode 100644 index 00000000000..785ded07f91 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsReplayPolicy.java @@ -0,0 +1,10 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; + +import com.fasterxml.jackson.annotation.JsonProperty; + +/** Replay policy for a NATS JetStream consumer. */ +@SuppressWarnings("unused") +public enum NatsReplayPolicy { + @JsonProperty("Instant") Instant, + @JsonProperty("Original") Original +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsUserAndPasswordConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsUserAndPasswordConfig.java new file mode 100644 index 00000000000..4addee557b2 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/NatsUserAndPasswordConfig.java @@ -0,0 +1,12 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; + +import com.fasterxml.jackson.annotation.JsonProperty; + +/** Username/password credentials for NATS authentication. */ +public class NatsUserAndPasswordConfig { + @JsonProperty("user") + public String user = ""; + + @JsonProperty("password") + public String password = ""; +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/ParquetEncoderConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/ParquetEncoderConfig.java new file mode 100644 index 00000000000..aa7cb14aa31 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/ParquetEncoderConfig.java @@ -0,0 +1,19 @@ +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; + +/** Configuration for the Parquet output format. */ +@SuppressWarnings("unused") +public class ParquetEncoderConfig implements IValidateConfig { + /** Number of records to buffer before writing a new Parquet file. */ + @JsonProperty("buffer_size_records") + public long bufferSizeRecords = 100_000; + + @Override + public boolean validate(ConfigReporter reporter) { + return true; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/ParquetParserConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/ParquetParserConfig.java new file mode 100644 index 00000000000..de7ab5c53b5 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/ParquetParserConfig.java @@ -0,0 +1,14 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; + +import org.dbsp.sqlCompiler.compiler.frontend.connectors.ConfigReporter; +import org.dbsp.sqlCompiler.compiler.frontend.connectors.IValidateConfig; + +/** Configuration for the Parquet input format. + * Validation still catches unknown fields. */ +@SuppressWarnings("unused") +public class ParquetParserConfig implements IValidateConfig { + @Override + public boolean validate(ConfigReporter reporter) { + return true; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PostgresCdcReaderConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PostgresCdcReaderConfig.java new file mode 100644 index 00000000000..63438035927 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PostgresCdcReaderConfig.java @@ -0,0 +1,40 @@ +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; + +/** Configuration for reading CDC data from Postgres logical replication. */ +@SuppressWarnings("unused") +public class PostgresCdcReaderConfig implements IValidateConfig { + @JsonProperty("uri") + public String uri = ""; + + @JsonProperty("publication") + public String publication = ""; + + @JsonProperty("source_table") + public String sourceTable = ""; + + // TLS fields (inlined from PostgresTlsConfig) + @Nullable @JsonProperty("ssl_ca_pem") public String sslCaPem = null; + @Nullable @JsonProperty("ssl_ca_location") public String sslCaLocation = null; + @Nullable @JsonProperty("ssl_client_pem") public String sslClientPem = null; + @Nullable @JsonProperty("ssl_client_location") public String sslClientLocation = null; + @Nullable @JsonProperty("ssl_client_key") public String sslClientKey = null; + @Nullable @JsonProperty("ssl_client_key_location") public String sslClientKeyLocation = null; + @Nullable @JsonProperty("ssl_certificate_chain_location") public String sslCertificateChainLocation = null; + @Nullable @JsonProperty("verify_hostname") public Boolean verifyHostname = null; + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + ok = ok && this.checkNonEmpty(reporter, uri, "uri"); + ok = ok && this.checkNonEmpty(reporter, publication, "publication"); + ok = ok && this.checkNonEmpty(reporter, sourceTable, "source_table"); + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PostgresReaderConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PostgresReaderConfig.java new file mode 100644 index 00000000000..d8e9d62ed41 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PostgresReaderConfig.java @@ -0,0 +1,36 @@ +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; + +/** Configuration for reading data from Postgres. */ +@SuppressWarnings("unused") +public class PostgresReaderConfig implements IValidateConfig { + @JsonProperty("uri") + public String uri = ""; + + @JsonProperty("query") + public String query = ""; + + // TLS fields (inlined from PostgresTlsConfig) + @Nullable @JsonProperty("ssl_ca_pem") public String sslCaPem = null; + @Nullable @JsonProperty("ssl_ca_location") public String sslCaLocation = null; + @Nullable @JsonProperty("ssl_client_pem") public String sslClientPem = null; + @Nullable @JsonProperty("ssl_client_location") public String sslClientLocation = null; + @Nullable @JsonProperty("ssl_client_key") public String sslClientKey = null; + @Nullable @JsonProperty("ssl_client_key_location") public String sslClientKeyLocation = null; + @Nullable @JsonProperty("ssl_certificate_chain_location") public String sslCertificateChainLocation = null; + @Nullable @JsonProperty("verify_hostname") public Boolean verifyHostname = null; + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + ok = ok && this.checkNonEmpty(reporter, uri, "uri"); + ok = ok && this.checkNonEmpty(reporter, query, "query"); + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PostgresWriteMode.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PostgresWriteMode.java new file mode 100644 index 00000000000..044e3d17f48 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PostgresWriteMode.java @@ -0,0 +1,7 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum PostgresWriteMode { + @JsonProperty("materialized") Materialized, + @JsonProperty("cdc") Cdc, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PostgresWriterConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PostgresWriterConfig.java new file mode 100644 index 00000000000..06972b8ca82 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PostgresWriterConfig.java @@ -0,0 +1,108 @@ +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; +import java.util.List; + +/** Configuration for the Postgres output connector. */ +@SuppressWarnings("unused") +public class PostgresWriterConfig implements IValidateConfig { + private static final String DEFAULT_CDC_OP_COLUMN = "__feldera_op"; + private static final String DEFAULT_CDC_TS_COLUMN = "__feldera_ts"; + + @JsonProperty("uri") + public String uri = ""; + + @JsonProperty("table") + public String table = ""; + + @JsonProperty("mode") + public PostgresWriteMode mode = PostgresWriteMode.Materialized; + + @JsonProperty("cdc_op_column") + public String cdcOpColumn = DEFAULT_CDC_OP_COLUMN; + + @JsonProperty("cdc_ts_column") + public String cdcTsColumn = DEFAULT_CDC_TS_COLUMN; + + // TLS fields (inlined from PostgresTlsConfig) + @Nullable @JsonProperty("ssl_ca_pem") public String sslCaPem = null; + @Nullable @JsonProperty("ssl_ca_location") public String sslCaLocation = null; + @Nullable @JsonProperty("ssl_client_pem") public String sslClientPem = null; + @Nullable @JsonProperty("ssl_client_location") public String sslClientLocation = null; + @Nullable @JsonProperty("ssl_client_key") public String sslClientKey = null; + @Nullable @JsonProperty("ssl_client_key_location") public String sslClientKeyLocation = null; + @Nullable @JsonProperty("ssl_certificate_chain_location") public String sslCertificateChainLocation = null; + @Nullable @JsonProperty("verify_hostname") public Boolean verifyHostname = null; + + @Nullable + @JsonProperty("max_records_in_buffer") + public Long maxRecordsInBuffer = null; + + @JsonProperty("max_buffer_size_bytes") + public long maxBufferSizeBytes = 1 << 20; + + @JsonProperty("on_conflict_do_nothing") + public boolean onConflictDoNothing = false; + + @JsonProperty("threads") + public int threads = 1; + + @Nullable + @JsonProperty("extra_columns") + public List extraColumns = null; + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + switch (mode) { + case Cdc -> { + if (cdcOpColumn.isBlank()) { + reporter.warnPath("cdc_op_column", "Invalid configuration", + "\"cdc_op_column\" cannot be empty in CDC mode"); + ok = false; + } else if (!cdcOpColumn.chars().allMatch(c -> c < 128)) { + reporter.warnPath("cdc_op_column", "Invalid configuration", + "\"cdc_op_column\" must contain only ASCII characters"); + ok = false; + } + if (cdcTsColumn.isBlank()) { + reporter.warnPath("cdc_ts_column", "Invalid configuration", + "\"cdc_ts_column\" cannot be empty in CDC mode"); + ok = false; + } else if (!cdcTsColumn.chars().allMatch(c -> c < 128)) { + reporter.warnPath("cdc_ts_column", "Invalid configuration", + "\"cdc_ts_column\" must contain only ASCII characters"); + ok = false; + } + if (onConflictDoNothing) { + reporter.warnPath("on_conflict_do_nothing", "Invalid configuration", + "\"on_conflict_do_nothing\" is not supported in CDC mode"); + ok = false; + } + } + case Materialized -> { + if (!cdcTsColumn.equals(DEFAULT_CDC_TS_COLUMN) && !cdcTsColumn.isBlank()) { + reporter.warnPath("cdc_ts_column", "Invalid configuration", + "\"cdc_ts_column\" must not be set in MATERIALIZED mode"); + ok = false; + } + if (!cdcOpColumn.equals(DEFAULT_CDC_OP_COLUMN) && !cdcOpColumn.isBlank()) { + reporter.warnPath("cdc_op_column", "Invalid configuration", + "\"cdc_op_column\" must not be set in MATERIALIZED mode"); + ok = false; + } + } + } + if (threads == 0) { + reporter.warnPath("threads", "Invalid configuration", + "\"threads\" must be at least 1"); + ok = false; + } + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PubSubInputConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PubSubInputConfig.java new file mode 100644 index 00000000000..ce6a778eeee --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/PubSubInputConfig.java @@ -0,0 +1,62 @@ +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; + +/** Configuration for the Google Pub/Sub input connector. */ +@SuppressWarnings("unused") +public class PubSubInputConfig implements IValidateConfig { + @Nullable + @JsonProperty("emulator") + public String emulator = null; + + @Nullable + @JsonProperty("credentials") + public String credentials = null; + + @Nullable + @JsonProperty("endpoint") + public String endpoint = null; + + @Nullable + @JsonProperty("pool_size") + public Integer poolSize = null; + + @Nullable + @JsonProperty("timeout_seconds") + public Integer timeoutSeconds = null; + + @Nullable + @JsonProperty("connect_timeout_seconds") + public Integer connectTimeoutSeconds = null; + + @Nullable + @JsonProperty("project_id") + public String projectId = null; + + @JsonProperty("subscription") + public String subscription = ""; + + @Nullable + @JsonProperty("snapshot") + public String snapshot = null; + + @Nullable + @JsonProperty("timestamp") + public String timestamp = null; + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + if (snapshot != null && timestamp != null) { + reporter.warnPath("snapshot", "Invalid configuration", + "\"snapshot\" and \"timestamp\" are mutually exclusive"); + ok = false; + } + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/RawParserConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/RawParserConfig.java new file mode 100644 index 00000000000..01691b14485 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/RawParserConfig.java @@ -0,0 +1,29 @@ +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; + +/** Configuration for the raw byte input format. */ +@SuppressWarnings("unused") +public class RawParserConfig implements IValidateConfig { + /** Whether to ingest each transport chunk as a single row ({@code blob}) + * or split on newlines ({@code lines}). */ + @JsonProperty("mode") + public RawParserMode mode = RawParserMode.Blob; + + /** Table column that will store the raw value. + * Required when the table has more than one column; that constraint + * cannot be checked here without schema context. */ + @Nullable + @JsonProperty("column_name") + public String columnName = null; + + @Override + public boolean validate(ConfigReporter reporter) { + return true; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/RawParserMode.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/RawParserMode.java new file mode 100644 index 00000000000..6075c05be80 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/RawParserMode.java @@ -0,0 +1,7 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum RawParserMode { + @JsonProperty("blob") Blob, + @JsonProperty("lines") Lines, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/RedisOutputConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/RedisOutputConfig.java new file mode 100644 index 00000000000..300d9cf34eb --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/RedisOutputConfig.java @@ -0,0 +1,21 @@ +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; + +/** Configuration for the Redis output connector. */ +@SuppressWarnings("unused") +public class RedisOutputConfig implements IValidateConfig { + @JsonProperty("connection_string") + public String connectionString = ""; + + @JsonProperty("key_separator") + public String keySeparator = ":"; + + @Override + public boolean validate(ConfigReporter reporter) { + return true; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/S3InputConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/S3InputConfig.java new file mode 100644 index 00000000000..e950f359252 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/S3InputConfig.java @@ -0,0 +1,63 @@ +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; + +/** Configuration for reading data from AWS S3. */ +@SuppressWarnings("unused") +public class S3InputConfig implements IValidateConfig { + @Nullable + @JsonProperty("aws_access_key_id") + public String awsAccessKeyId = null; + + @Nullable + @JsonProperty("aws_secret_access_key") + public String awsSecretAccessKey = null; + + @JsonProperty("no_sign_request") + public boolean noSignRequest = false; + + @Nullable + @JsonProperty("key") + public String key = null; + + @Nullable + @JsonProperty("prefix") + public String prefix = null; + + @JsonProperty("region") + public String region = ""; + + @JsonProperty("bucket_name") + public String bucketName = ""; + + @Nullable + @JsonProperty("endpoint_url") + public String endpointUrl = null; + + @JsonProperty("max_concurrent_fetches") + public int maxConcurrentFetches = 8; + + @JsonProperty("max_retries") + public int maxRetries = 5; + + @Override + public boolean validate(ConfigReporter reporter) { + boolean ok = true; + if (region.isBlank()) { + reporter.warnPath("region", "Invalid configuration", + "required field \"region\" is missing or empty"); + ok = false; + } + if (bucketName.isBlank()) { + reporter.warnPath("bucket_name", "Invalid configuration", + "required field \"bucket_name\" is missing or empty"); + ok = false; + } + return ok; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/SubjectNameStrategy.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/SubjectNameStrategy.java new file mode 100644 index 00000000000..7ca68285c2a --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/SubjectNameStrategy.java @@ -0,0 +1,8 @@ +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; +import com.fasterxml.jackson.annotation.JsonProperty; + +public enum SubjectNameStrategy { + @JsonProperty("topic_name") TopicName, + @JsonProperty("record_name") RecordName, + @JsonProperty("topic_record_name") TopicRecordName, +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/UrlInputConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/UrlInputConfig.java new file mode 100644 index 00000000000..d2f3b0ee8e7 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/UrlInputConfig.java @@ -0,0 +1,21 @@ +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; + +/** Configuration for reading data from an HTTP or HTTPS URL. */ +@SuppressWarnings("unused") +public class UrlInputConfig implements IValidateConfig { + @JsonProperty("path") + public String path = ""; + + @JsonProperty("pause_timeout") + public int pauseTimeout = 60; + + @Override + public boolean validate(ConfigReporter reporter) { + return true; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/package-info.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/package-info.java new file mode 100644 index 00000000000..72cc7e9d1ea --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/package-info.java @@ -0,0 +1,15 @@ +/** + * POJO configuration classes and their helper enums for connector format and transport configs. + * These classes are used using Jackson to automatically deserialize connector configurations + * to check for errors. + */ + +@ParametersAreNonnullByDefault +@FieldsAreNonnullByDefault +@MethodsAreNonnullByDefault +package org.dbsp.sqlCompiler.compiler.frontend.connectors.config; + +import org.dbsp.util.FieldsAreNonnullByDefault; +import org.dbsp.util.MethodsAreNonnullByDefault; + +import javax.annotation.ParametersAreNonnullByDefault; diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/package-info.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/package-info.java new file mode 100644 index 00000000000..631d05f5725 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/package-info.java @@ -0,0 +1,13 @@ +/** + * Validation of connector configurations in SQL. + */ + +@ParametersAreNonnullByDefault +@FieldsAreNonnullByDefault +@MethodsAreNonnullByDefault +package org.dbsp.sqlCompiler.compiler.frontend.connectors; + +import org.dbsp.util.FieldsAreNonnullByDefault; +import org.dbsp.util.MethodsAreNonnullByDefault; + +import javax.annotation.ParametersAreNonnullByDefault; diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/ConnectorTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/ConnectorTests.java new file mode 100644 index 00000000000..ff06d06c63d --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/ConnectorTests.java @@ -0,0 +1,1456 @@ +package org.dbsp.sqlCompiler.compiler.sql; + +import org.dbsp.sqlCompiler.compiler.DBSPCompiler; +import org.dbsp.sqlCompiler.compiler.TestUtil; +import org.dbsp.sqlCompiler.compiler.sql.tools.BaseSQLTests; +import org.junit.Assert; +import org.junit.Test; + +import java.util.stream.Collectors; + +/** Tests for validation of connector format and transport configs. */ +public class ConnectorTests extends BaseSQLTests { + /** Compiles SQL, asserts exit code 0, and verifies each message appears in compiler output. */ + private void runConnectorTest(String sql, String... expectedMessages) { + DBSPCompiler compiler = this.chattyCompiler(); + compiler.submitStatementsForCompilation(sql); + compiler.getFinalCircuit(true); + Assert.assertEquals(0, compiler.messages.exitCode); + for (String msg : expectedMessages) + TestUtil.assertMessagesContain(compiler.messages, msg); + } + + /** Adds four spaces of indentation to every line of {@code body}. */ + private static String indent4(String body) { + return body.lines() + .map(line -> " " + line) + .collect(Collectors.joining("\n")); + } + + /** + * Compiles a table input connector with the given connector body (the JSON object + * content after {@code "name": "c",}), asserts exit code 0, and verifies each + * expected message appears in compiler output. + * + *

The generated SQL wraps the body as: + * 

+     * CREATE TABLE T (x INT) WITH (    -- line 1
+     *   'connectors' = '[{             -- line 2
+     *     "name": "c",                 -- line 3
+     *     {connectorBody}              -- lines 4+
+     *   }]'
+     * );
+     *
+ * For a typical {@code "format": {"name": "X", "config": {F}}} body, the first field + * {@code F} is on line 7 with 8 spaces of indentation. + */ + private void tableConnectorTest(String connectorBody, String... expectedMessages) { + runConnectorTest( + "CREATE TABLE T (x INT) WITH (\n" + + " 'connectors' = '[{\n" + + " \"name\": \"c\",\n" + + indent4(connectorBody) + "\n" + + " }]'\n" + + ");", + expectedMessages); + } + + /** + * Compiles a view output connector with the given connector body, asserts exit code 0, + * and verifies each expected message appears in compiler output. + * + *

The generated SQL wraps the body as: + * 

+     * CREATE TABLE T (x INT);          -- line 1
+     * CREATE VIEW V WITH (             -- line 2
+     *   'connectors' = '[{             -- line 3
+     *     "name": "c",                 -- line 4
+     *     {connectorBody}              -- lines 5+
+     *   }]'
+     * ) AS SELECT * FROM T;
+     *
+ * For a typical {@code "format": {"name": "X", "config": {F}}} body, the first field + * {@code F} is on line 8 with 8 spaces of indentation. + */ + private void viewConnectorTest(String connectorBody, String... expectedMessages) { + runConnectorTest( + "CREATE TABLE T (x INT);\n" + + "CREATE VIEW V WITH (\n" + + " 'connectors' = '[{\n" + + " \"name\": \"c\",\n" + + indent4(connectorBody) + "\n" + + " }]'\n" + + ") AS SELECT * FROM T;", + expectedMessages); + } + + // ---- Connector structure validation ---- + + @Test + public void formatNotObject() { + runConnectorTest(""" + CREATE TABLE T (x INT) WITH ( + 'connectors' = '[{ + "name": "c", + "format": "csv" + }]' + );""", + "\"format\" must be a JSON object"); + } + + @Test + public void formatMissingName() { + runConnectorTest(""" + CREATE TABLE T (x INT) WITH ( + 'connectors' = '[{ + "name": "c", + "format": {} + }]' + );""", + "\"format\" must have a \"name\" field"); + } + + @Test + public void formatNameNotString() { + runConnectorTest(""" + CREATE TABLE T (x INT) WITH ( + 'connectors' = '[{ + "name": "c", + "format": {"name": 42} + }]' + );""", + "\"format.name\" must be a string"); + } + + @Test + public void formatConfigNotObject() { + runConnectorTest(""" + CREATE TABLE T (x INT) WITH ( + 'connectors' = '[{ + "name": "c", + "format": {"name": "csv", "config": "bad"} + }]' + );""", + "\"format.config\" must be a JSON object"); + } + + @Test + public void transportNotObject() { + runConnectorTest(""" + CREATE TABLE T (x INT) WITH ( + 'connectors' = '[{ + "name": "c", + "transport": "kafka" + }]' + );""", + "\"transport\" must be a JSON object"); + } + + @Test + public void transportMissingName() { + runConnectorTest(""" + CREATE TABLE T (x INT) WITH ( + 'connectors' = '[{ + "name": "c", + "transport": {} + }]' + );""", + "\"transport\" must have a \"name\" field"); + } + + @Test + public void transportNameNotString() { + runConnectorTest(""" + CREATE TABLE T (x INT) WITH ( + 'connectors' = '[{ + "name": "c", + "transport": {"name": 42} + }]' + );""", + "\"transport.name\" must be a string"); + } + + @Test + public void transportConfigNotObject() { + runConnectorTest(""" + CREATE TABLE T (x INT) WITH ( + 'connectors' = '[{ + "name": "c", + "transport": {"name": "file_input", "config": "bad"} + }]' + );""", + "\"transport.config\" must be a JSON object"); + } + + @Test + public void issue4896() { + // An unnamed connector should produce a warning. + String sql = """ + CREATE TABLE T (COL1 INT) WITH ( + 'connectors' = '[{ + "url": "localhost" + }]' + );"""; + DBSPCompiler compiler = this.chattyCompiler(); + compiler.submitStatementsForCompilation(sql); + compiler.getFinalCircuit(true); + TestUtil.assertMessagesContain(compiler.messages, + "warning: Unnamed connector: Connector nr. 1 for table 't' does not have a name.\n" + + "It is recommended to name all connectors using the \"name\" property; " + + "names will be required in the future."); + + // A non-string connector name should be a compilation error. + sql = """ + CREATE TABLE T (COL1 INT) WITH ( + 'connectors' = '[{ + "name": [], + "url": "localhost" + }]' + );"""; + this.statementsFailingInCompilation(sql, """ + Compilation error: Expected a string value for the connector "name" property + 3| "name": [], + ^ + 4| "url": "localhost" + """); + + // Duplicate connector names on the same table should be a compilation error. + sql = """ + CREATE TABLE T (COL1 INT) WITH ( + 'connectors' = '[{ + "name": "Bob", + "url": "localhost" + }, { + "name": "Bob", + "url": "localhost:8080" + }]' + );"""; + this.statementsFailingInCompilation(sql, """ + error: Compilation error: Two connectors for the same table 't' cannot have the same name: 'Bob' + 6| "name": "Bob", + ^ + 7| "url": "localhost:8080\""""); + } + + @Test + public void testPreprocessorValidation() { + // Missing "name" field in preprocessor. + this.statementsFailingInCompilation(""" + CREATE TABLE T(x INT) WITH ('connectors' = '[{ + "name": "0", + "transport": { + "name": "datagen", + "config": {} + }, + "preprocessor": [{ + "config": {} + }] + }]');""", """ + Compilation error: Preprocessor must have a field "name" + 7| "preprocessor": [{ + ^ + 8| "config": {}"""); + // Missing "message_oriented" field. + this.statementsFailingInCompilation(""" + CREATE TABLE T(x INT) WITH ('connectors' = '[{ + "name": "0", + "transport": { + "name": "datagen", + "config": {} + }, + "preprocessor": [{ + "name": "Bob", + "config": {} + }] + }]');""", """ + Compilation error: Preprocessor must have a field "message_oriented" + 7| "preprocessor": [{ + ^ + 8| "name": "Bob","""); + // Wrong type for "message_oriented". + this.statementsFailingInCompilation(""" + CREATE TABLE T(x INT) WITH ('connectors' = '[{ + "name": "0", + "transport": { + "name": "datagen", + "config": {} + }, + "preprocessor": [{ + "message_oriented": "streaming", + "name": "Bob", + "config": {} + }] + }]');""", """ + Compilation error: Preprocessor field "message_oriented" must be a Boolean value + 8| "message_oriented": "streaming", + ^ + 9| "name": "Bob","""); + } + + @Test + public void testPostprocessorValidation() { + // Missing "name" field in postprocessor. + this.statementsFailingInCompilation(""" + CREATE TABLE T(x INT); + CREATE VIEW V WITH ('connectors' = '[{ + "name": "0", + "postprocessor": [{ + "config": {} + }] + }]') AS SELECT * FROM T;""", """ + Compilation error: Postprocessor must have a field "name" + 4| "postprocessor": [{ + ^ + 5| "config": {}"""); + } + + // ---- CSV format config ---- + + @Test + public void csvValidConfig() { + tableConnectorTest(""" + "format": { + "name": "csv", + "config": { + "delimiter": ";", + "headers": true, + "double_quote": false, + "flexible": false, + "trim": "fields" + } + }"""); + } + + @Test + public void csvNonAsciiDelimiter() { + tableConnectorTest(""" + "format": { + "name": "csv", + "config": { + "delimiter": "α" + } + }""", + "field \"delimiter\" must be an ASCII character", + " 7| \"delimiter\": \"α\"\n" + + " ^"); + } + + @Test + public void csvSameDelimiterAndQuote() { + tableConnectorTest(""" + "format": { + "name": "csv", + "config": { + "delimiter": ";", + "quote": ";" + } + }""", + "fields \"delimiter\" and \"quote\" must be different characters", + " 7| \"delimiter\": \";\",\n" + + " ^"); + } + + @Test + public void csvUnknownField() { + tableConnectorTest(""" + "format": { + "name": "csv", + "config": { + "delimter": ";" + } + }""", + "warning: Invalid configuration: unknown field \"delimter\"\n" + + " 7| \"delimter\": \";\"\n" + + " ^"); + } + + @Test + public void csvWrongTypeForBoolean() { + tableConnectorTest(""" + "format": { + "name": "csv", + "config": { + "headers": "yes" + } + }""", + "Cannot deserialize value of type `boolean` from String \"yes\"", + " 7| \"headers\": \"yes\"\n" + + " ^"); + } + + @Test + public void csvInvalidTrimValue() { + tableConnectorTest(""" + "format": { + "name": "csv", + "config": { + "trim": "both" + } + }""", + "field \"trim\": invalid value \"both\"; valid values are: \"none\", \"headers\", \"fields\", \"all\"", + " 7| \"trim\": \"both\"\n" + + " ^"); + } + + // ---- CSV encoder (output) config ---- + + @Test + public void csvEncoderValidConfig() { + viewConnectorTest(""" + "format": { + "name": "csv", + "config": { + "delimiter": ";", + "buffer_size_records": 500 + } + }"""); + } + + @Test + public void csvEncoderUnknownField() { + viewConnectorTest(""" + "format": { + "name": "csv", + "config": { + "delimter": ";" + } + }""", + "warning: Invalid configuration: unknown field \"delimter\"\n" + + " 8| \"delimter\": \";\"\n" + + " ^"); + } + + @Test + public void csvEncoderNonAsciiDelimiter() { + viewConnectorTest(""" + "format": { + "name": "csv", + "config": { + "delimiter": "α" + } + }""", + "field \"delimiter\" must be an ASCII character", + " 8| \"delimiter\": \"α\"\n" + + " ^"); + } + + // ---- JSON format config ---- + + @Test + public void jsonParserUnknownField() { + tableConnectorTest(""" + "format": { + "name": "json", + "config": { + "update_format": "raw", + "streem": true + } + }""", + "warning: Invalid configuration: unknown field \"streem\"\n" + + " 8| \"streem\": true\n" + + " ^"); + } + + @Test + public void jsonParserInvalidUpdateFormat() { + tableConnectorTest(""" + "format": { + "name": "json", + "config": { + "update_format": "upsert" + } + }""", + "field \"update_format\": invalid value \"upsert\"; valid values are:", + " 7| \"update_format\": \"upsert\"\n" + + " ^"); + } + + @Test + public void jsonEncoderKeyFieldsWithoutDebezium() { + viewConnectorTest(""" + "format": { + "name": "json", + "config": { + "update_format": "raw", + "key_fields": ["x"] + } + }""", + "\"key_fields\" is only valid with \"update_format\": \"debezium\""); + } + + // ---- Raw format config ---- + + @Test + public void rawParserInvalidMode() { + tableConnectorTest(""" + "format": { + "name": "raw", + "config": { + "mode": "chunks" + } + }""", + "field \"mode\": invalid value \"chunks\"; valid values are: \"blob\", \"lines\"", + " 7| \"mode\": \"chunks\"\n" + + " ^"); + } + + // ---- Parquet format config ---- + + @Test + public void parquetParserUnknownField() { + tableConnectorTest(""" + "format": { + "name": "parquet", + "config": { + "compress": true + } + }""", + "warning: Invalid configuration: unknown field \"compress\"\n" + + " 7| \"compress\": true\n" + + " ^"); + } + + @Test + public void parquetEncoderUnknownField() { + viewConnectorTest(""" + "format": { + "name": "parquet", + "config": { + "compress": true + } + }""", + "warning: Invalid configuration: unknown field \"compress\"\n" + + " 8| \"compress\": true\n" + + " ^"); + } + + // ---- Avro format config ---- + + @Test + public void avroParserUnknownField() { + tableConnectorTest(""" + "format": { + "name": "avro", + "config": { + "update_format": "debezium", + "compress": true + } + }""", + "warning: Invalid configuration: unknown field \"compress\""); + } + + @Test + public void avroParserSchemaAndRegistryMutuallyExclusive() { + tableConnectorTest(""" + "format": { + "name": "avro", + "config": { + "schema": "{}", + "registry_urls": ["http://localhost:8081"] + } + }""", + "\"schema\" and \"registry_urls\" are mutually exclusive"); + } + + @Test + public void avroEncoderCdcFieldRequiresRaw() { + viewConnectorTest(""" + "format": { + "name": "avro", + "config": { + "update_format": "debezium", + "cdc_field": "op" + } + }""", + "\"cdc_field\" is only valid with \"update_format\": \"raw\""); + } + + // ---- Delta Table transport config ---- + + @Test + public void deltaReaderMissingMode() { + tableConnectorTest(""" + "transport": { + "name": "delta_table_input", + "config": { + "uri": "s3://bucket/table" + } + }""", + "required field \"mode\" is missing"); + } + + @Test + public void deltaReaderVersionAndDatetimeExclusive() { + tableConnectorTest(""" + "transport": { + "name": "delta_table_input", + "config": { + "uri": "s3://bucket/table", + "mode": "snapshot", + "version": 5, + "datetime": "2024-01-01T00:00:00Z" + } + }""", + "\"version\" and \"datetime\" are mutually exclusive"); + } + + @Test + public void deltaReaderCdcFieldsOutsideCdcMode() { + tableConnectorTest(""" + "transport": { + "name": "delta_table_input", + "config": { + "uri": "s3://bucket/table", + "mode": "snapshot", + "cdc_delete_filter": "col IS NULL" + } + }""", + "\"cdc_delete_filter\" is only valid with \"mode\": \"cdc\""); + } + + @Test + public void deltaWriterInvalidLogRetention() { + viewConnectorTest(""" + "transport": { + "name": "delta_table_output", + "config": { + "uri": "s3://bucket/table", + "log_retention_duration": "30 days" + } + }""", + "invalid \"log_retention_duration\" value \"30 days\": " + + "expected format \"interval \""); + } + + @Test + public void deltaWriterZeroThreads() { + viewConnectorTest(""" + "transport": { + "name": "delta_table_output", + "config": { + "uri": "s3://bucket/table", + "threads": 0 + } + }""", + "\"threads\" must be greater than 0"); + } + + // ---- HTTP transport config ---- + + @Test + public void httpOutputUnknownField() { + viewConnectorTest(""" + "transport": { + "name": "http_output", + "config": { + "backpressure": true, + "bufffer_size": 100 + } + }""", + "warning: Invalid configuration: unknown field \"bufffer_size\"\n" + + " 9| \"bufffer_size\": 100\n" + + " ^"); + } + + // ---- S3 transport config ---- + + @Test + public void s3InputMissingRegion() { + tableConnectorTest(""" + "transport": { + "name": "s3_input", + "config": { + "bucket_name": "my-bucket" + } + }""", + "required field \"region\" is missing or empty"); + } + + @Test + public void s3InputMissingBucketName() { + tableConnectorTest(""" + "transport": { + "name": "s3_input", + "config": { + "region": "us-east-1" + } + }""", + "required field \"bucket_name\" is missing or empty"); + } + + @Test + public void s3InputUnknownField() { + tableConnectorTest(""" + "transport": { + "name": "s3_input", + "config": { + "region": "us-east-1", + "bucket_name": "my-bucket", + "acl": "public-read" + } + }""", + "unknown field \"acl\""); + } + + @Test + public void s3InputValidConfig() { + tableConnectorTest(""" + "transport": { + "name": "s3_input", + "config": { + "region": "us-east-1", + "bucket_name": "my-bucket", + "prefix": "data/", + "max_retries": 3 + } + }"""); + } + + // ---- Postgres transport config ---- + + @Test + public void postgresReaderUnknownField() { + tableConnectorTest(""" + "transport": { + "name": "postgres_input", + "config": { + "uri": "postgres://localhost/db", + "query": "SELECT * FROM t", + "batch_size": 1000 + } + }""", + "unknown field \"batch_size\""); + } + + @Test + public void postgresCdcReaderUnknownField() { + tableConnectorTest(""" + "transport": { + "name": "postgres_cdc_input", + "config": { + "uri": "postgres://localhost/db", + "publication": "my_pub", + "source_table": "public.orders", + "batch_size": 1000 + } + }""", + "unknown field \"batch_size\""); + } + + @Test + public void postgresWriterCdcMissingOpColumn() { + viewConnectorTest(""" + "transport": { + "name": "postgres_output", + "config": { + "uri": "postgres://localhost/db", + "table": "t", + "mode": "cdc", + "cdc_op_column": "" + } + }""", + "\"cdc_op_column\" cannot be empty in CDC mode"); + } + + @Test + public void postgresWriterCdcOnConflictDoNothing() { + viewConnectorTest(""" + "transport": { + "name": "postgres_output", + "config": { + "uri": "postgres://localhost/db", + "table": "t", + "mode": "cdc", + "on_conflict_do_nothing": true + } + }""", + "\"on_conflict_do_nothing\" is not supported in CDC mode"); + } + + @Test + public void postgresWriterMaterializedCdcTsColumn() { + viewConnectorTest(""" + "transport": { + "name": "postgres_output", + "config": { + "uri": "postgres://localhost/db", + "table": "t", + "cdc_ts_column": "my_ts" + } + }""", + "\"cdc_ts_column\" must not be set in MATERIALIZED mode"); + } + + @Test + public void postgresWriterZeroThreads() { + viewConnectorTest(""" + "transport": { + "name": "postgres_output", + "config": { + "uri": "postgres://localhost/db", + "table": "t", + "threads": 0 + } + }""", + "\"threads\" must be at least 1"); + } + + @Test + public void postgresWriterUnknownField() { + viewConnectorTest(""" + "transport": { + "name": "postgres_output", + "config": { + "uri": "postgres://localhost/db", + "table": "t", + "batch_size": 1000 + } + }""", + "unknown field \"batch_size\""); + } + + // ---- Kafka transport config ---- + + @Test + public void kafkaInputMissingTopic() { + tableConnectorTest(""" + "transport": { + "name": "kafka_input", + "config": { + "bootstrap.servers": "localhost:9092" + } + }""", + "required field \"topic\" is missing or empty"); + } + + @Test + public void kafkaInputInvalidLogLevel() { + tableConnectorTest(""" + "transport": { + "name": "kafka_input", + "config": { + "topic": "my-topic", + "log_level": "verbose" + } + }""", + "field \"log_level\": invalid value \"verbose\""); + } + + @Test + public void kafkaOutputMissingTopic() { + viewConnectorTest(""" + "transport": { + "name": "kafka_output", + "config": { + "bootstrap.servers": "localhost:9092" + } + }""", + "required field \"topic\" is missing or empty"); + } + + @Test + public void kafkaOutputInvalidLogLevel() { + viewConnectorTest(""" + "transport": { + "name": "kafka_output", + "config": { + "topic": "my-topic", + "log_level": "verbose" + } + }""", + "field \"log_level\": invalid value \"verbose\""); + } + + // ---- Iceberg transport config ---- + + @Test + public void icebergReaderMissingMode() { + tableConnectorTest(""" + "transport": { + "name": "iceberg_input", + "config": { + "metadata_location": "s3://bucket/table/metadata/v1.json" + } + }""", + "required field \"mode\" is missing"); + } + + @Test + public void icebergReaderSnapshotIdAndDatetimeExclusive() { + tableConnectorTest(""" + "transport": { + "name": "iceberg_input", + "config": { + "mode": "snapshot", + "metadata_location": "s3://bucket/table/metadata/v1.json", + "snapshot_id": 42, + "datetime": "2024-01-01T00:00:00Z" + } + }""", + "\"snapshot_id\" and \"datetime\" are mutually exclusive"); + } + + @Test + public void icebergReaderMissingLocation() { + tableConnectorTest(""" + "transport": { + "name": "iceberg_input", + "config": { + "mode": "snapshot" + } + }""", + "missing metadata location"); + } + + @Test + public void icebergReaderMetadataLocationAndCatalogTypeExclusive() { + tableConnectorTest(""" + "transport": { + "name": "iceberg_input", + "config": { + "mode": "snapshot", + "metadata_location": "s3://bucket/table/metadata/v1.json", + "catalog_type": "glue", + "table_name": "db.orders", + "glue.warehouse": "s3://warehouse/" + } + }""", + "\"metadata_location\" is not supported when \"catalog_type\" is set"); + } + + @Test + public void icebergReaderMissingTableName() { + tableConnectorTest(""" + "transport": { + "name": "iceberg_input", + "config": { + "mode": "snapshot", + "catalog_type": "glue", + "glue.warehouse": "s3://warehouse/" + } + }""", + "\"table_name\" must be specified when \"catalog_type\" is set"); + } + + @Test + public void icebergReaderGlueMissingWarehouse() { + tableConnectorTest(""" + "transport": { + "name": "iceberg_input", + "config": { + "mode": "snapshot", + "catalog_type": "glue", + "table_name": "db.orders" + } + }""", + "missing Iceberg warehouse location"); + } + + @Test + public void icebergReaderGluePropWithoutGlueCatalog() { + tableConnectorTest(""" + "transport": { + "name": "iceberg_input", + "config": { + "mode": "snapshot", + "metadata_location": "s3://bucket/table/metadata/v1.json", + "glue.warehouse": "s3://warehouse/" + } + }""", + "\"glue.warehouse\" is only valid when \"catalog_type\": \"glue\""); + } + + @Test + public void icebergReaderRestMissingUri() { + tableConnectorTest(""" + "transport": { + "name": "iceberg_input", + "config": { + "mode": "snapshot", + "catalog_type": "rest", + "table_name": "db.orders" + } + }""", + "missing Iceberg REST catalog URI"); + } + + @Test + public void icebergReaderInvalidMode() { + tableConnectorTest(""" + "transport": { + "name": "iceberg_input", + "config": { + "mode": "batch", + "metadata_location": "s3://bucket/table/metadata/v1.json" + } + }""", + "field \"mode\": invalid value \"batch\""); + } + + // ---- File transport config ---- + + @Test + public void fileInputUnknownField() { + tableConnectorTest(""" + "transport": { + "name": "file_input", + "config": { + "path": "/data/input.csv", + "encoding": "utf-8" + } + }""", + "unknown field \"encoding\""); + } + + @Test + public void fileInputValidConfig() { + tableConnectorTest(""" + "transport": { + "name": "file_input", + "config": { + "path": "/data/input.csv", + "follow": true + } + }"""); + } + + @Test + public void fileOutputUnknownField() { + viewConnectorTest(""" + "transport": { + "name": "file_output", + "config": { + "path": "/data/output.csv", + "encoding": "utf-8" + } + }""", + "unknown field \"encoding\""); + } + + // ---- URL transport config ---- + + @Test + public void urlInputUnknownField() { + tableConnectorTest(""" + "transport": { + "name": "url_input", + "config": { + "path": "https://example.com/data.csv", + "retries": 3 + } + }""", + "unknown field \"retries\""); + } + + @Test + public void urlInputValidConfig() { + tableConnectorTest(""" + "transport": { + "name": "url_input", + "config": { + "path": "https://example.com/data.csv", + "pause_timeout": 30 + } + }"""); + } + + // ---- Pub/Sub transport config ---- + + @Test + public void pubSubInputSnapshotAndTimestampExclusive() { + tableConnectorTest(""" + "transport": { + "name": "pub_sub_input", + "config": { + "subscription": "my-sub", + "snapshot": "my-snapshot", + "timestamp": "2024-08-17T16:39:57-08:00" + } + }""", + "\"snapshot\" and \"timestamp\" are mutually exclusive"); + } + + @Test + public void pubSubInputUnknownField() { + tableConnectorTest(""" + "transport": { + "name": "pub_sub_input", + "config": { + "subscription": "my-sub", + "max_messages": 100 + } + }""", + "unknown field \"max_messages\""); + } + + @Test + public void pubSubInputValidConfig() { + tableConnectorTest(""" + "transport": { + "name": "pub_sub_input", + "config": { + "subscription": "my-sub", + "project_id": "my-project" + } + }"""); + } + + // ---- DynamoDB transport config ---- + + @Test + public void dynamodbOutputValidConfig() { + viewConnectorTest(""" + "transport": { + "name": "dynamodb_output", + "config": { + "table": "my-table", + "region": "us-east-1" + } + }"""); + } + + @Test + public void dynamodbOutputTableTooShort() { + viewConnectorTest(""" + "transport": { + "name": "dynamodb_output", + "config": { + "table": "ab", + "region": "us-east-1" + } + }""", + "DynamoDB requires 3–255 characters"); + } + + @Test + public void dynamodbOutputTableInvalidChars() { + viewConnectorTest(""" + "transport": { + "name": "dynamodb_output", + "config": { + "table": "my/table", + "region": "us-east-1" + } + }""", + "contains invalid characters"); + } + + @Test + public void dynamodbOutputMissingRegion() { + viewConnectorTest(""" + "transport": { + "name": "dynamodb_output", + "config": { + "table": "my-table" + } + }""", + "\"region\" cannot be empty"); + } + + @Test + public void dynamodbOutputBatchSizeExceedsLimit() { + viewConnectorTest(""" + "transport": { + "name": "dynamodb_output", + "config": { + "table": "my-table", + "region": "us-east-1", + "write_mode": "batch", + "batch_size": 26 + } + }""", + "\"batch_size\" must be between 1 and 25 for \"batch\" write mode"); + } + + @Test + public void dynamodbOutputCredentialsMustBePaired() { + viewConnectorTest(""" + "transport": { + "name": "dynamodb_output", + "config": { + "table": "my-table", + "region": "us-east-1", + "aws_access_key_id": "AKIA..." + } + }""", + "\"aws_access_key_id\" and \"aws_secret_access_key\" must be specified together"); + } + + @Test + public void dynamodbOutputZeroThreads() { + viewConnectorTest(""" + "transport": { + "name": "dynamodb_output", + "config": { + "table": "my-table", + "region": "us-east-1", + "threads": 0 + } + }""", + "\"threads\" must be at least 1"); + } + + @Test + public void dynamodbOutputUnknownField() { + viewConnectorTest(""" + "transport": { + "name": "dynamodb_output", + "config": { + "table": "my-table", + "region": "us-east-1", + "consistant_reads": true + } + }""", + "unknown field \"consistant_reads\""); + } + + // ---- Redis transport config ---- + + @Test + public void redisOutputUnknownField() { + viewConnectorTest(""" + "transport": { + "name": "redis_output", + "config": { + "connection_string": "redis://localhost:6379/0", + "ttl_seconds": 3600 + } + }""", + "unknown field \"ttl_seconds\""); + } + + @Test + public void redisOutputValidConfig() { + viewConnectorTest(""" + "transport": { + "name": "redis_output", + "config": { + "connection_string": "redis://localhost:6379/0", + "key_separator": "|" + } + }"""); + } + + // ---- Clock transport config ---- + + @Test + public void clockInputZeroResolution() { + tableConnectorTest(""" + "transport": { + "name": "clock", + "config": { + "clock_resolution_usecs": 0 + } + }""", + "\"clock_resolution_usecs\" must be greater than 0"); + } + + @Test + public void clockInputUnknownField() { + tableConnectorTest(""" + "transport": { + "name": "clock", + "config": { + "clock_resolution_usecs": 1000, + "tick_rate": 60 + } + }""", + "unknown field \"tick_rate\""); + } + + @Test + public void clockInputValidConfig() { + tableConnectorTest(""" + "transport": { + "name": "clock", + "config": { + "clock_resolution_usecs": 1000, + "http_driven": true + } + }"""); + } + + // ---- NATS transport config ---- + + @Test + public void natsInputValidConfig() { + tableConnectorTest(""" + "transport": { + "name": "nats_input", + "config": { + "connection_config": { + "server_url": "nats://localhost:4222" + }, + "stream_name": "my-stream", + "consumer_config": { + "deliver_policy": "All" + } + } + }"""); + } + + @Test + public void natsInputValidConfigByStartSequence() { + tableConnectorTest(""" + "transport": { + "name": "nats_input", + "config": { + "connection_config": { + "server_url": "nats://localhost:4222", + "connection_timeout_secs": 30 + }, + "stream_name": "orders", + "inactivity_timeout_secs": 120, + "consumer_config": { + "deliver_policy": {"ByStartSequence": {"start_sequence": 42}}, + "replay_policy": "Original", + "filter_subjects": ["orders.>"] + } + } + }"""); + } + + @Test + public void natsInputMissingStreamName() { + tableConnectorTest(""" + "transport": { + "name": "nats_input", + "config": { + "connection_config": { + "server_url": "nats://localhost:4222" + }, + "consumer_config": { + "deliver_policy": "All" + } + } + }""", + "required field \"stream_name\" is missing or empty"); + } + + @Test + public void natsInputMissingConnectionConfig() { + tableConnectorTest(""" + "transport": { + "name": "nats_input", + "config": { + "stream_name": "my-stream", + "consumer_config": { + "deliver_policy": "All" + } + } + }""", + "required field \"connection_config\" is missing"); + } + + @Test + public void natsInputMissingServerUrl() { + tableConnectorTest(""" + "transport": { + "name": "nats_input", + "config": { + "connection_config": { + "connection_timeout_secs": 10 + }, + "stream_name": "my-stream", + "consumer_config": { + "deliver_policy": "All" + } + } + }""", + "required field \"connection_config.server_url\" is missing or empty"); + } + + @Test + public void natsInputMissingDeliverPolicy() { + tableConnectorTest(""" + "transport": { + "name": "nats_input", + "config": { + "connection_config": { + "server_url": "nats://localhost:4222" + }, + "stream_name": "my-stream", + "consumer_config": { + "replay_policy": "Instant" + } + } + }""", + "required field \"consumer_config.deliver_policy\" is missing"); + } + + @Test + public void natsInputInvalidReplayPolicy() { + tableConnectorTest(""" + "transport": { + "name": "nats_input", + "config": { + "connection_config": { + "server_url": "nats://localhost:4222" + }, + "stream_name": "my-stream", + "consumer_config": { + "deliver_policy": "All", + "replay_policy": "fast" + } + } + }""", + "invalid value \"fast\""); + } + + @Test + public void natsInputUnknownFieldTopLevel() { + tableConnectorTest(""" + "transport": { + "name": "nats_input", + "config": { + "connection_config": { + "server_url": "nats://localhost:4222" + }, + "stream_name": "my-stream", + "consumer_config": { + "deliver_policy": "All" + }, + "max_reconnects": 5 + } + }""", + "unknown field \"max_reconnects\""); + } + + @Test + public void natsInputUnknownFieldInConnectionConfig() { + tableConnectorTest(""" + "transport": { + "name": "nats_input", + "config": { + "connection_config": { + "server_url": "nats://localhost:4222", + "tls_enabled": true + }, + "stream_name": "my-stream", + "consumer_config": { + "deliver_policy": "All" + } + } + }""", + "unknown field \"tls_enabled\""); + } + + @Test + public void natsInputZeroInactivityTimeout() { + tableConnectorTest(""" + "transport": { + "name": "nats_input", + "config": { + "connection_config": { + "server_url": "nats://localhost:4222" + }, + "stream_name": "my-stream", + "inactivity_timeout_secs": 0, + "consumer_config": { + "deliver_policy": "All" + } + } + }""", + "\"inactivity_timeout_secs\" must be at least 1"); + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/MetadataTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/MetadataTests.java index 6b8ca07ab63..f6fa57c1981 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/MetadataTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/MetadataTests.java @@ -400,54 +400,6 @@ FOREIGN KEY (cc_num) REFERENCES CUSTOMER(cc_num) Assert.assertFalse(rust.contains(CreateTableStatement.CONNECTORS)); } - @Test - public void issue4896() { - String sql = """ - CREATE TABLE T (COL1 INT) WITH ( - 'connectors' = '[{ - "url": "localhost" - }]' - );"""; - DBSPCompiler compiler = this.chattyCompiler(); - compiler.submitStatementsForCompilation(sql); - // Force compilation - compiler.getFinalCircuit(true); - Assert.assertTrue(compiler.messages.toString().contains( - "warning: Unnamed connector: Connector nr. 1 for table 't' does not have a name.\n" + - "It is recommended to name all connectors using the \"name\" property; " + - "names will be required in the future.")); - - sql = """ - CREATE TABLE T (COL1 INT) WITH ( - 'connectors' = '[{ - "name": [], - "url": "localhost" - }]' - );"""; - this.statementsFailingInCompilation(sql, """ - Compilation error: Expected a string value for the connector "name" property - 3| "name": [], - ^ - 4| "url": "localhost" - """); - - sql = """ - CREATE TABLE T (COL1 INT) WITH ( - 'connectors' = '[{ - "name": "Bob", - "url": "localhost" - }, { - "name": "Bob", - "url": "localhost:8080" - }]' - );"""; - this.statementsFailingInCompilation(sql,""" - error: Compilation error: Two connectors for the same table 't' cannot have the same name: 'Bob' - 6| "name": "Bob", - ^ - 7| "url": "localhost:8080\""""); - } - @Test public void stripConnectors() throws IOException, SQLException { // Test that the connectors are stripped from the generated Rust @@ -976,58 +928,6 @@ pub fn i128_sum_post(val: i128_sum_accumulator_type) -> ByteArray { } } - @Test - public void testPreprocessorValidation() { - this.statementsFailingInCompilation(""" - CREATE TABLE T(x INT) WITH ('connectors' = '[{ - "name": "0", - "transport": { - "name": "datagen", - "config": {} - }, - "preprocessor": [{ - "config": {} - }] - }]');""", """ - Compilation error: Preprocessor must have a field "name" - 7| "preprocessor": [{ - ^ - 8| "config": {}"""); - this.statementsFailingInCompilation(""" - CREATE TABLE T(x INT) WITH ('connectors' = '[{ - "name": "0", - "transport": { - "name": "datagen", - "config": {} - }, - "preprocessor": [{ - "name": "Bob", - "config": {} - }] - }]');""", """ - Compilation error: Preprocessor must have a field "message_oriented" - 7| "preprocessor": [{ - ^ - 8| "name": "Bob","""); - this.statementsFailingInCompilation(""" - CREATE TABLE T(x INT) WITH ('connectors' = '[{ - "name": "0", - "transport": { - "name": "datagen", - "config": {} - }, - "preprocessor": [{ - "message_oriented": "streaming", - "name": "Bob", - "config": {} - }] - }]');""", """ - Compilation error: Preprocessor field "message_oriented" must be a Boolean value - 8| "message_oriented": "streaming", - ^ - 9| "name": "Bob","""); - } - @Test public void testUDP() throws IOException, InterruptedException, SQLException { // Test user-defined preprocessor @@ -1112,22 +1012,6 @@ fn create( } } - @Test - public void testPostprocessorValidation() { - this.statementsFailingInCompilation(""" - CREATE TABLE T(x INT); - CREATE VIEW V WITH ('connectors' = '[{ - "name": "0", - "postprocessor": [{ - "config": {} - }] - }]') AS SELECT * FROM T;""", """ - Compilation error: Postprocessor must have a field "name" - 4| "postprocessor": [{ - ^ - 5| "config": {}"""); - } - @Test public void testPostprocessor() throws IOException, InterruptedException, SQLException { // Test user-defined postprocessor