jonathansudhakar1 opened a new issue, #41017:
URL: https://github.com/apache/beam/issues/41017
### What happened?
A Beam SQL Parquet table can write a `DATE` column, but reading it back
fails. On master (74ab567d2e3), `SELECT *` throws a `DateTimeException`:
```
java.time.DateTimeException: Invalid value for EpochDay (valid values
-365243219162 - 365241780471): 1752451200000
at java.time.LocalDate.ofEpochDay(LocalDate.java:345)
at
org.apache.beam.sdk.extensions.sql.impl.rel.BeamCalcRel.toBeamObject(BeamCalcRel.java:443)
```
Projecting the column (`SELECT d FROM t` on a table `t (id INTEGER, d
DATE)`) throws a `ClassCastException` instead:
```
java.lang.ClassCastException: class org.joda.time.Instant cannot be cast to
class java.time.LocalDate
at
org.apache.beam.sdk.schemas.logicaltypes.Date.toBaseType(Date.java:35)
```
Minimal repro, run as a JUnit test in `sdks/java/extensions/sql` with the
DirectRunner (`writePipeline` and `readPipeline` are `TestPipeline`s):
```java
BeamSqlEnv env = BeamSqlEnv.inMemory(new ParquetTableProvider());
env.executeDdl("CREATE EXTERNAL TABLE t (d DATE) TYPE parquet LOCATION
'/tmp/date-table/'");
BeamSqlRelUtils.toPCollection(
writePipeline, env.parseQuery("INSERT INTO t VALUES (DATE
'2025-07-14')"));
writePipeline.run().waitUntilFinish(); // OK: d is written as an Avro int
(date)
BeamSqlRelUtils.toPCollection(readPipeline, env.parseQuery("SELECT * FROM
t"));
readPipeline.run().waitUntilFinish(); // DateTimeException above
```
1752451200000 is 2025-07-14T00:00Z in epoch **millis**, while
`LocalDate.ofEpochDay` expects epoch days.
**Cause.** The read path gives SQL rows whose schema is inferred from Avro,
not the table's schema:
- `ParquetTable.buildIOReader` reads with
`ParquetIO.read(schema).withBeamSchemas(true)` and then applies
`Convert.toRows()`
([L62-L63](https://github.com/apache/beam/blob/74ab567d2e389b61dec063839325bf88bab4484e/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTable.java#L62-L63),
[L71-L77](https://github.com/apache/beam/blob/74ab567d2e389b61dec063839325bf88bab4484e/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/parquet/ParquetTable.java#L71-L77)).
- With `withBeamSchemas(true)`, ParquetIO's coder is
`AvroUtils.schemaCoder(avroSchema)` ([ParquetIO
L730-L732](https://github.com/apache/beam/blob/74ab567d2e389b61dec063839325bf88bab4484e/sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetIO.java#L730-L732)).
That infers the Beam schema from Avro, and an Avro `date` becomes `DATETIME`
([AvroUtils
L1058-L1059](https://github.com/apache/beam/blob/74ab567d2e389b61dec063839325bf88bab4484e/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java#L1058-L1059)).
Its value is a Joda `Instant` ([`convertDateStrict`,
L1665-L1668](https://github.com/apache/beam/blob/74ab567d2e389b61dec063839325bf88bab4484e/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java#L1665-L1668)).
- The table schema says the column is
`FieldType.logicalType(SqlTypes.DATE)`, backed by `java.time.LocalDate`
([CalciteUtils
L111](https://github.com/apache/beam/blob/74ab567d2e389b61dec063839325bf88bab4484e/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/utils/CalciteUtils.java#L111)).
So SQL receives an `Instant` where it expects a `LocalDate`.
The write side was fixed in #39414, so only reads are affected now. In the
same temporary test, every other scalar type round-trips through `INSERT` and
`SELECT *`: TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DECIMAL,
BOOLEAN, TIMESTAMP, CHAR, VARCHAR, BINARY and VARBINARY.
**Expected:** `SELECT` on a DATE column returns the `LocalDate` that was
written.
**Possible fix** (I'd like a maintainer's view before sending one, since
`AvroUtils` is shared by many IOs):
1. In `AvroUtils`, let the strict Avro-to-Beam conversion turn a `date`
value into a `LocalDate` when the target field type is `SqlTypes.DATE`. Today
`convertDateStrict` rejects every target except `DATETIME`. This is additive:
callers that infer the schema from Avro still get `DATETIME`.
2. In `ParquetTable`, convert the records with the table's own Beam schema,
or its projected subset, instead of the Avro-inferred one. For example, use
`ParquetIO.read(schema)` without `withBeamSchemas`, followed by `MapElements`
with `AvroUtils.getGenericRecordToRowFunction(tableSchema)` and
`setRowSchema(tableSchema)`.
Changing `AvroUtils.toBeamSchema` to infer `SqlTypes.DATE` from Avro `date`
would also work, but it changes behavior for every Avro-schema user, so I'd
rather not.
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [ ] Component: Python SDK
- [x] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Prism Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
