Repository navigation
fix(flink): run the engine on flink - #46
Conversation
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 3581eda6fa
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| val counted = plan(batchEnv()) | ||
| .flatMap(perRecord, pairType) | ||
| .keyBy(new FirstField, Types.STRING) | ||
| .sum(1) |
There was a problem hiding this comment.
When the same contract message or quality-check key occurs on multiple records, KeyedStream.sum emits rolling intermediate totals rather than one final total—the Flink 1.18 API explicitly describes this as a rolling sum. Consequently two occurrences can be collected as (key, 1) and (key, 2), producing duplicate violations, and the collection limit can be exhausted before other failed checks or final totals appear. Use a bounded aggregation that emits one final count per key before applying the limit.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Not the case in batch mode. Every job here sets RuntimeExecutionMode.BATCH, and a keyed sum in batch mode emits one final value per key rather than a running total per element. That was the reason for the mode, and the reason is now in a test: "a violation many records share is reported once, with the number of records behind it" runs validate over two records that break the contract the same way and asserts one violation with recordsAffected 2. Rolling totals would answer (name is Bob, 1) and (name is Bob, 2) and fail it.
| def decode[A](json: String, decoder: DataDecoder[A]): Option[A] = | ||
| decoder | ||
| .decode(EncodedData(json.getBytes(StandardCharsets.UTF_8), DataFormat.JSON), DataFormat.JSON) | ||
| .toOption |
There was a problem hiding this comment.
For CSV and JSONL sources, a valid DataDecoder[A] may support only the declared source format, but every row is passed to it as DataFormat.JSON. In particular, a CSV-only decoder—which the previous parity fixture used—rejects every converted row, leaving data empty and causing filters, maps, validation, and writes to silently drop the entire dataset while metadata still reports records. Preserve the row format in the plan or reject unsupported codecs instead of unconditionally invoking them as JSON.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
This is engine-wide, not Flink-specific: SparkFrameOps decodes every row as JSON too, including rows read from a CSV file, because JSON is the one representation the codecs and the engines both speak. Making only Flink reject a codec would make the two engines disagree on the same source, which is the thing this design avoids. Recorded as a known limit in docs/plan/v1.0-readiness.md instead, including that supportsFormat is not consulted so the caller gets no signal.
| val sampleRows = FlinkStreamOps.collect(plan, sampleSize) | ||
| val sample = sampleRows.flatMap(row => FlinkRows.decode(row, decoder)) | ||
| val recordCount = | ||
| if (sampleRows.sizeIs < sampleSize) sampleRows.size.toLong else FlinkStreamOps.count(plan) |
There was a problem hiding this comment.
When any source row is malformed or otherwise rejected by DataDecoder[A], sample drops it but recordCount still counts the raw plan row. This makes count/isEmpty disagree with the records in Dataset[A], inflates contract-score denominators, and lets write report more records than its re-encoding plan actually emits. Derive the count from successfully decoded rows, as the in-memory implementation does.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Same answer: the Spark engine takes recordCount from df.count(), which also counts rows rather than records that decoded, so Flink counting raw rows is what keeps the two engines agreeing. The in-memory engine is the one that differs. Recorded as a known limit in docs/plan/v1.0-readiness.md, including that it inflates the denominator validate scores against.
| def count(plan: FlinkPlan): Long = { | ||
| val rows = plan(batchEnv()).executeAndCollect() | ||
| try rows.asScala.length.toLong |
There was a problem hiding this comment.
For every plan with at least 100 rows, this polls every record back to the submitting process merely to count it; Flink documents that executeAndCollect transfers stream events through the REST API to the application process. Thus each read/filter/map of a large dataset incurs full-dataset network transfer, and Scala's Iterator.length is only an Int, so counts above Int.MaxValue can overflow before conversion to Long. Perform a Flink-side Long aggregation and collect only its single result.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in 310502b. count now maps each row to a (key, 1) pair under a single key, sums in the job, and collects one result, so no row crosses to the driver to be counted and the sum is a Long throughout.
| def csvLineToJson(header: List[String], line: String): Option[String] = { | ||
| val trimmed = line.trim | ||
| if (trimmed.isEmpty || trimmed == header.mkString(",")) None | ||
| else { |
There was a problem hiding this comment.
The header is removed by comparing every line's contents with the normalized header text, so a legitimate later record equal to the column names—for example a name,role row in a file whose header is also name,role—is silently discarded. Conversely, a header containing spaces such as id, name is normalized to id,name, fails this comparison, and is counted as data. Track and skip the first nonblank physical line instead of filtering rows by value.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in 310502b. The driver now passes the header line as the file holds it, so the comparison is against that line rather than a line rebuilt from trimmed names: a header written "id, name" is recognised. Test: "a header the file pads with spaces is still a header". A record whose text equals the header line is still dropped with it; a line-oriented source reports no line number, so there is nothing to tell them apart by, and that is in the scaladoc.
The Flink module was pinned to Scala 2.12 for the deprecated Flink Scala API, so it did not resolve, and its algebra forwarded every call to the in-memory engine. It now builds on 2.13 against Flink's Java API and runs read, write, filter, map, validate and the quality checks as batch jobs over every record.