andygrove opened a new issue, #2555:
URL: https://github.com/apache/datafusion-ballista/issues/2555
**Is your feature request related to a problem or challenge? Please describe
what you are trying to do.**
Ballista doesn't discover table partitions itself. Tables are registered on
the client through DataFusion's `SessionContext`, and the scheduler prunes by
whatever partition columns the resulting `ListingTable` declares. So Ballista
only finds Hive-style partitions (`d=2024-01-01/`) when DataFusion does, and
DataFusion's discovery does less than Spark's:
- `register_parquet` never discovers partitions. The caller has to pass
`table_partition_cols`.
- `CREATE EXTERNAL TABLE ... LOCATION` without a column list does discover
them (`datafusion.execution.listing_table_factory_infer_partitions`, on by
default), with two gaps:
1. Every partition column becomes `Dictionary(UInt16, Utf8)`. A `DATE`
filter still prunes, through a cast, but the column is no longer a date.
2. It fails when the files also store the partition column.
`ListingTable::try_new` appends the partition fields to the file schema without
removing duplicates, so the first query fails with a `DuplicateQualifiedField`
schema error. Data written by DataFusion with
`datafusion.execution.keep_partition_by_columns = true` has this layout.
Spark's `spark.read.parquet` discovers partitions by default, infers their
types (dates, integers and so on), and accepts a partition column that also
appears in the files. On data like this, Ballista users get either no partition
pruning, a missing partition column, or a schema error, depending on the API
and the layout. The `tpch` benchmark runner needed its own detection to work
around this (#2553).
To reproduce, write a small table partitioned by `d` twice. In the first
copy the files don't store `d`, and in the second they do:
```sh
datafusion-cli \
-c "COPY (SELECT * FROM (VALUES (1, 10, DATE '2024-01-01'), (2, 20, DATE
'2024-01-02'), (3, 30, DATE '2024-01-03')) AS v(id, amount, d)) TO
'/tmp/hive-repro/path_only/' STORED AS PARQUET PARTITIONED BY (d);" \
-c "SET datafusion.execution.keep_partition_by_columns = true;" \
-c "COPY (SELECT * FROM (VALUES (1, 10, DATE '2024-01-01'), (2, 20, DATE
'2024-01-02'), (3, 30, DATE '2024-01-03')) AS v(id, amount, d)) TO
'/tmp/hive-repro/stored/' STORED AS PARQUET PARTITIONED BY (d);"
```
Then, in `ballista-cli` in standalone mode (`main` at `60cd1a52f`,
DataFusion 55.1.0):
```sql
CREATE EXTERNAL TABLE path_only STORED AS PARQUET LOCATION
'/tmp/hive-repro/path_only/';
SELECT DISTINCT arrow_typeof(d) FROM path_only;
-- Dictionary(UInt16, Utf8)
SELECT * FROM path_only WHERE d >= DATE '2024-01-02';
-- Returns ids 2 and 3. EXPLAIN shows the filter as
-- CAST(path_only.d AS Date32) >= Date32("2024-01-02"), and only the
-- d=2024-01-02 and d=2024-01-03 files are scanned.
CREATE EXTERNAL TABLE stored STORED AS PARQUET LOCATION
'/tmp/hive-repro/stored/';
SELECT * FROM stored;
-- SchemaError(DuplicateQualifiedField { qualifier: Bare { table: "stored"
}, name: "d" }, Some(""))
```
Declaring the columns by hand works, and keeps `d` a `Date32`:
```sql
CREATE EXTERNAL TABLE stored_declared (id BIGINT, amount BIGINT, d DATE)
STORED AS PARQUET PARTITIONED BY (d) LOCATION '/tmp/hive-repro/stored/';
SELECT * FROM stored_declared WHERE d >= DATE '2024-01-02';
-- Returns ids 2 and 3.
```
**Describe the solution you'd like**
Partition discovery that matches what Spark users expect:
1. When a discovered partition column is also stored in the files, keep one
column instead of failing, typed as the files store it.
2. Infer partition column types (dates, integers, decimals) from the
directory values, as Spark does with
`spark.sql.sources.partitionColumnTypeInference.enabled`, so that date filters
compare dates. This changes column types, so it may need a setting.
3. Consider an opt-in for discovering partitions in `register_parquet` and
the other `register_*` methods too.
The discovery code lives in DataFusion (`ListingTableFactory::create`,
`ListingTableConfig::infer_partitions_from_path` and `ListingTable::try_new`),
so the natural home for 1 and 2 is upstream. Ballista would pick them up with
its next DataFusion upgrade, and the `tpch` runner's workaround from #2553
could then be removed.
**Describe alternatives you've considered**
- Ballista could register its own `TableProviderFactory` for `PARQUET` in
client sessions (ballista-cli, Python and the Rust client) that does
Spark-style discovery. That would fix `CREATE EXTERNAL TABLE` without waiting
for a DataFusion release. It wouldn't change `register_parquet`, though, and it
duplicates logic that belongs in DataFusion.
- Declaring partition columns by hand, as above. That works today, but users
have to know to do it, and Spark users don't expect to need it.
**Additional context**
- `DESCRIBE` doesn't work in Ballista yet (`LogicalPlan serde is not yet
implemented for DescribeTable`), so the repro uses `arrow_typeof` to show the
column type.
- #2553 adds partition detection to the `tpch` runner as a workaround.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
