andygrove opened a new issue, #2580: URL: https://github.com/apache/datafusion-ballista/issues/2580
**Is your feature request related to a problem or challenge? Please describe what you are trying to do.** Most data that Spark users would bring to Ballista is stored as Hive tables. Those are Parquet or ORC files in `key=value/` partition directories on S3 or HDFS, usually described by a Hive Metastore or the AWS Glue Data Catalog. Spark reads them by name. It discovers the partition columns and infers their types, reads `__HIVE_DEFAULT_PARTITION__` as NULL, and resolves column names case-insensitively. Ballista doesn't discover tables itself. Tables are registered on the client through DataFusion, so Ballista reads Hive tables only as well as DataFusion's `ListingTable` does, and there's no metastore catalog. A Spark user has to declare every table, its partition columns and their types by hand. Even then, some Spark-written layouts fail or return wrong results. For example, the `tpcds` benchmark needed a view to read the NULL partitions in Spark-written TPC-DS data (#2565). This epic collects the gaps. Some already have their own issues, such as #2555, which covers partition discovery and type inference in detail, so this epic links them rather than repeating them. Most of the fixes belong in DataFusion, and Ballista picks them up when it upgrades. The Ballista-specific work is making catalogs and table providers work across the client, scheduler and executors, and documenting a Spark-compatible setup. **Describe the solution you'd like** Partitioned directories - [ ] Discover partition columns when registering Parquet tables, keep a partition column that the files also store, and infer partition column types as Spark does (#2555) - [ ] Read `__HIVE_DEFAULT_PARTITION__` as NULL instead of failing to cast it (apache/datafusion#18083) - [ ] Ignore hidden files and directories (names starting with `_` or `.`), as Hadoop and Spark do. Hive writes data files without an extension (`000000_0`). Reading those today means dropping the `.parquet` extension filter, and then every other file in the table directory is read as data too. - [ ] Support partitions that aren't `key=value` directories under the table root. A metastore can point a partition anywhere (`ALTER TABLE ... ADD PARTITION ... LOCATION`), but `ListingTable` only takes partition values from paths under the table root and skips other files. Catalogs - [ ] A Hive Metastore catalog that resolves databases, tables and partitions by name (apache/datafusion#2209) - [ ] An AWS Glue Data Catalog, which Spark on EMR and Glue uses - [ ] Make catalog-backed tables work from every client (Rust, Python and `ballista-cli`) and plan on the scheduler. A table provider other than `ListingTable` needs a logical extension codec on both the client and the scheduler. File formats - [ ] ORC, which is common in Hive deployments (apache/datafusion#4707, `datafusion-orc` in datafusion-contrib) - [ ] Hive text tables (`\001` field delimiter and `\N` for NULL) Spark-written Parquet - [ ] INT96 timestamps. Spark writes timestamps as INT96 by default, and DataFusion reads them as nanoseconds unless `datafusion.execution.parquet.coerce_int96` is set, so dates outside 1677 to 2262 overflow. - [ ] Case-insensitive column names. Spark resolves `customerid` against a `CustomerId` field. DataFusion lowercases unquoted identifiers and then matches case-sensitively, so queries on mixed-case schemas fail unless every name is quoted. - [ ] A documented Spark-compatible configuration, possibly as part of the existing `spark-compat` feature, that sets `coerce_int96`, `binary_as_string` and so on Writing - [ ] Write NULL partition keys to `__HIVE_DEFAULT_PARTITION__`. DataFusion 55 writes a NULL integer key to `k=0` and a NULL string key to `k=`, so those rows read back as `0` and `''`. DataFusion `main` now rejects NULL partition keys instead (apache/datafusion#24951, not released yet), and writing them as `__HIVE_DEFAULT_PARTITION__` is part of apache/datafusion#18083. - [ ] Check that partition values are escaped and unescaped the way Hive does it, so Spark can read tables that Ballista writes and the other way round - [ ] `INSERT OVERWRITE` with Spark's static and dynamic partition overwrite modes Later - [ ] Bucketed tables. Read Hive and Spark bucketing metadata and use it to skip shuffles for joins and aggregations. Note that Spark and Hive use different bucket hash functions. **Describe alternatives you've considered** - Ask users to convert their tables to Iceberg or Delta first. Many are migrating anyway, and Iceberg support is tracked in #890. But that turns leaving Spark into a two-step migration, and Hive tables are where most users start. - Keep declaring tables by hand with `CREATE EXTERNAL TABLE ... PARTITIONED BY`. That works for simple layouts but doesn't scale to hundreds of tables, and it doesn't help with NULL partitions or partitions outside the table root. **Additional context** Related: #2299 (roadmap discussion), #2249 (Spark Connect) and the `spark-compat` feature, which registers Spark-compatible functions from `datafusion-spark`. -- 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]
