Repository navigation
Conversation
kellen
changed the title
labianchin
reviewed
Oct 22, 2026
labianchin
left a comment
labianchin
left a comment
Collaborator
There was a problem hiding this comment.
-
Oops, I ended up rebasing #1040. Can you rebase or may I?
-
Perhaps split out the dbeam-core Avro changes into their own PR against master. They seem to change default behaviour for existent avro users. I think we can review and ship those changes independently.
About this PR, I've noticed a few changes:
-
array element types other than int/float/bool/bytea/string (_numeric, _timestamp, _date, _json) now fail, where #1040 wrote them as strings
-
NULL array items now fail by default
-
the footer schema drops dbeam's per-field properties
-
Is the idea to be deliberate about failure? Traditionally we'd favor some sane fallback (e.g. to string) because people might evolve schema with new fields and we'd rather avoid unnecessary failures .
When --useTimestampMicros is set (with --useAvroLogicalTypes), TIMESTAMP, DATE, TIME and TIME_WITH_TIMEZONE columns are written as INT64 TIMESTAMP(MICROS, isAdjustedToUTC=true) with microsecond precision taken from java.sql.Timestamp nanos, matching dbeam-core's Avro semantics. The footer/_AVRO_SCHEMA.avsc Avro schema now uses timestamp-micros to match, replacing the previous workaround that always disabled micros. The writer picks the unit from the target Parquet field's logical type, so values also match a user-provided --parquetSchemaFilePath. Default (MILLIS) behaviour is unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Adds --useLogicalTypes to JdbcExportPipelineOptions and keeps --useAvroLogicalTypes as a deprecated, backward-compatible alias. Both options are now nullable with no default; JdbcExportArgsFactory resolves the effective value (whichever is set, false by default) and fails with an IllegalArgumentException if both are set and disagree. The --useTimestampMicros check now names --useLogicalTypes. Documents --useTimestampMicros in docs/type-conversion.md and README, and fixes the Avro doc table to say timestamp-millis (not time-millis). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
When --parquetSchemaFilePath is supplied, reject TIMESTAMP fields whose unit disagrees with --useTimestampMicros, so the written values and the footer Avro schema / _AVRO_SCHEMA.avsc cannot contradict each other. Add H2 round-trip tests reading output through the footer Avro schema with AvroParquetReader (MICROS and MILLIS), and clarify the changelog. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The parquet.avro.schema footer key and _AVRO_SCHEMA.avsc were built from the JDBC metadata by dbeam-core's JdbcAvroSchema, independently of the Parquet schema, and could contradict the written columns: LIST items were non-null while Parquet elements are optional, typed_first_row element types came from the first row instead of the type name, a NULL first-row ARRAY dropped the footer, and --parquetSchemaFilePath field names/types were ignored. Add ParquetAvroSchema, an in-module Parquet->Avro converter following parquet-avro's AvroSchemaConverter, and derive the footer from the MessageType used for writing, keeping dbeam's namespace, doc and column properties. Add AvroParquetReader round-trip tests and an e2e footer vs _AVRO_SCHEMA.avsc check. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…rter Replace the hand-written Parquet-to-Avro converter with AvroSchemaConverter (parquet.avro.add-list-element-records=false) and only set the top-level record name, namespace and doc. Drop record/field props and field docs. parquet-avro moves to compile scope. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Parquet list elements are always optional, so the flag has no effect. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The Parquet writers take column types from JDBC and write fields by position, so a user-provided schema can only rename fields or change annotations safely. Fail fast with a FailedValidationException listing every offending column when the field count, primitive/group shape, primitive type, fixed length, repetition or LIST annotation differs, or when a logical type annotation would change how written bytes are read. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…cros Fold checkTimestampUnits into ParquetSchemaValidator: the expected schema is derived with the job's --useTimestampMicros value, so TIMESTAMP(MICROS,true) is accepted exactly with the flag and TIMESTAMP(MILLIS,true) exactly without it; NANOS and non-UTC are rejected with a FailedValidationException. The error output shows the schema for the job's actual flags. Validation uses the single LIMIT 1 query on the schema-file path; the footer Avro schema is still derived from the write schema. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…a annotation The writer no longer inspects the target field's TIMESTAMP annotation; the flag is threaded through JdbcParquetArgs to the write support. The validator still rejects annotations that contradict the flag, so unannotated int64 schema-file fields are written in the flag's unit. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…seTimestampMicros Also document that the flag alone decides the unit for Avro and Parquet exports. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Fail fast in JdbcParquetJob.create, before args parsing or any output, with an IllegalArgumentException pointing to --parquetSchemaFilePath. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Schema-file-vs-data mismatches are validation failures: both the Avro --avroSchemaFilePath check and the Parquet --parquetSchemaFilePath validator now fail with FailedValidationException (exit code 50). The Avro check moves from JdbcExportArgsFactory.fromPipelineOptions to JdbcAvroJob.prepareExport (still before any output). Pure CLI option conflicts remain IllegalArgumentException (exit code 43). Document the exit-code mapping in the README. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The Avro writer binary-encodes values by field position from the JDBC column types, with union index 0 for NULL and 1 otherwise, while a schema file is used verbatim as the writer schema. Any mismatch silently produces files that cannot be read correctly. JdbcAvroJob.prepareExport now runs one LIMIT 1 query, derives the expected schema with JdbcAvroSchema (with and without logical types) and checks the schema file with the new AvroSchemaValidator: field count, ["null", T] unions, Avro types incl. array items and their nullability, and logical types (absent or the one dbeam would generate). Renames and doc/namespace changes are allowed. The timestamp-unit check and its messages are kept and run first. Failures are FailedValidationException. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…roCodec --parquetCodec now defaults to snappy and is validated at the very start of prepareExport, before any metadata file is written, with the list of supported values. zstd1 .. zstd22 select a Zstandard compression level (parquet.compression.codec.zstd.level). --avroCodec is no longer mapped; a non-default value logs a warning (an explicit --avroCodec=deflate6 cannot be distinguished from the default). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The metric was reported after each query with the cumulative position of the output stream, which double-counted when one writer handles several queries and missed the last row group and the footer (often reporting just the 4-byte magic). Report the file size once, after parquetWriter.close(). Beam closes writers in @FinishBundle, where the DirectRunner has no current metrics container, so fall back to the container captured while writing. dbeam-core's Avro writer still reports the cumulative size per query (over-counting with several queries per writer); left unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Match dbeam-core's JdbcAvroSchema, which fails schema generation for these types with the same messages, instead of silently exporting them as strings. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Use try-with-resources for the per-query PreparedStatement and the --preCommand Statement, close the connection when creating the ParquetWriter or running a query fails (Beam then discards the writer via cleanup() without calling finishWrite), and close the connection in finishWrite even if closing the ParquetWriter fails. --preCommand still runs before every query, matching JdbcAvroIO. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Opt-in (default false, requires --useLogicalTypes): DATE is written as int32 (DATE), TIME/TIME_WITH_TIMEZONE as TIME(MILLIS|MICROS, true) per --useTimestampMicros, and DECIMAL/NUMERIC with a JDBC precision of 1..38 (0 <= scale <= precision) as fixed_len_byte_array(n) (DECIMAL(p,s)). Unbounded precisions (PostgreSQL NUMERIC reports 0 or 131089) and precisions above 38 keep the string encoding. Values are rescaled to the column scale; values that need rounding or exceed the precision fail. DECIMAL uses fixed_len_byte_array for all precisions instead of int32/int64 for small ones: parquet-avro maps int32/int64 DECIMAL to plain Avro int/long, which would drop the scale from the footer Avro schema read by AvroParquetReader. Schema files may use these annotations only with the flag. Default behavior is unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Field names replace characters outside [A-Za-z0-9_] with '_', so columns like a-b and a_b map to the same field. JdbcParquetSchema silently built a schema with duplicate names and JdbcAvroSchema failed with Avro's generic duplicate-field error. Both now fail listing the colliding source column names. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The Parquet suite now mirrors the Avro runs (codecs via --parquetCodec, parallel queries, array modes) and adds --useLogicalTypes (uuid, timestamp-millis), --useTimestampMicros, --parquetNativeTypes and --dataOnly. Each run is checked by the new e2e/VerifyParquetOutput.java, which reads all .parquet files of all run_N directories through the footer Avro schema with AvroParquetReader: rows per run, row_number coverage, logical types, timestamp units and demo_table values, and that the footer equals _AVRO_SCHEMA.avsc (or that no metadata files exist with --dataOnly). The timestamp-micros check also reads all files. parquet-tools, whose failures were masked by '|| echo skipping', is dropped from e2e.sh and from the CI setup. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The Avro writer reported the cumulative output size after each query, so a writer handling several queries over-counted. Report the file size once, after dataFileWriter.close() (last block flushed), falling back to the metrics container captured while writing because Beam closes writers in @FinishBundle where the DirectRunner has no current container. Same approach as the Parquet writer. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Typed Parquet arrays defaulted unknown element types (timestamp, date, numeric, json, ...) to binary (STRING) and wrote them with toString(). Fail schema generation instead, naming the column and element type and suggesting --arrayMode=bytes. Integer, floating point, boolean and bytea elements stay typed; text, varchar, bpchar/char, name and uuid stay STRING. H2-style names (SMALLINT/DOUBLE PRECISION/CHARACTER VARYING/UUID ARRAY, ...) resolve to the same element types; unknown ones are rejected. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Avro: a nested array element ended in the generic "Value of type ... is not supported" error. Parquet typed arrays threw ClassCastException for numeric/bool/bytea elements and wrote "[Ljava.lang.String;@..." strings for text/uuid elements. Both now fail with "Column 'x' contains a multidimensional array, which is not supported; use --arrayMode=bytes ...". JDBC metadata does not expose the dimensions (int4[][] is _int4), so this is detected on the first such value; Parquet checks all elements before writing anything of the value. bytea[] elements (byte[]) are still written; bytea[][] (byte[][] elements) is rejected. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
ArrayHandlingMode.validateValue threw a plain RuntimeException, which mapped to the generic exit code 49. Throw IllegalArgumentException with the allowed values, like other invalid-argument checks. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Match the Avro export: typed LIST elements are required by default and
optional with --nullableArrayItems. Without the flag a NULL array item
fails with the Avro export's message ("Array item is null in column 'x',
use --nullableArrayItems") before any part of the value is written.
The footer Avro schema follows (array<T> vs array<["null",T]>), and
--parquetSchemaFilePath element repetition must agree with the flag.
Removes the "--nullableArrayItems has no effect" warning.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
kellen
force-pushed
the
parquet-timestamp-micros
branch
from
October 23, 2026 08:39
353ded8 to
a392f4a
Compare
…allback The footer parquet.avro.schema / _AVRO_SCHEMA.avsc now carries the same record properties (tableName, connectionUrl) and field docs and properties (columnName, sqlCode, typeName, columnClassName, columnTypeName) as dbeam's Avro export. They are taken from the JDBC metadata by column position, so --parquetSchemaFilePath field names are kept. Previously the derived schema had none of them. Typed ARRAY element types without a Parquet mapping (numeric, timestamp, date, json, ...) still fail by default. The new --parquetUnsupportedArrayItemsAsString flag writes them as BINARY (STRING) with each element's toString() and logs a warning per column. The writer only falls back to toString() when the schema element is STRING, which only the flag produces (schema files are validated against the same derived schema). Docs: clarify the footer schema wording and document both changes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Collaborator
Author
|
(1) Done. (2) They change default behavior but mostly buggy behavior, so I think it's OK to keep here unless you object.
(3) This matches Avro. Users can use (4) This also matches Avro: items are non-nullable unless (5) Fixed: the footer schema now keeps dbeam's record and field properties and docs, matched to columns by position. (6) Yes; the old fallback was silent. Added |
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Follow-ups to #1040: closes its known limitations and the gaps found in review, and brings the Parquet and Avro exports to parity.
New options
--useLogicalTypes: format-neutral;--useAvroLogicalTypesstays as a deprecated alias (conflicting values fail).--useTimestampMicrosfor Parquet: TIMESTAMP/DATE/TIME columns asTIMESTAMP(MICROS), sub-ms precision kept. The flag alone decides the unit in both formats.--parquetCodec: defaults tosnappy(was gzip via--avroCodec=deflate6); validated before any output;zstd1..zstd22set the level.--parquetNativeTypes(off by default): native Parquet DATE, TIME and DECIMAL; DECIMAL falls back to string when precision is unknown or > 38.Correctness
parquet.avro.schema/_AVRO_SCHEMA.avscderived from the written Parquet schema via parquet-avro'sAvroSchemaConverter(lists unwrapped). Previously built from JDBC metadata and could contradict the data (array nullability, element types, schema files).bytesWrittencounted once per file after close, in both Avro and Parquet (previously undercounted / double-counted).