Opt-in schema merging across files for Parquet/ORC TVFs

[jiangxt2]

I'd like to discuss optional schema merging across files when querying Parquet or ORC directly through Doris file TVFs.

Use case and scope

The use case is ad hoc SQL over an evolving file directory, without first creating a table or registering a table schema. Newer files may contain a business field absent from historical files. Users may want the query to expose that field and fill missing values with NULL.

This proposal applies only to direct file TVFs, with an initial scope limited to supported scalar fields at the top level. It does not change the existing schema-management behavior of Doris tables or catalog tables. The TVF's query schema would be resolved before execution and remain fixed for that query.

I contributed Ray Data's opt-in full schema inference for Parquet and ORC, both merged. I also developed the independent open-source ray-doris connector, which enables Ray Data to read Doris data directly.

In a supporting Ray DataSourceV2 build, merge_schema=True inspects all eligible files; the default uses bounded sampling. Doris file TVFs use the first file's schema by default. More precisely, the 4.1.5 selection path picks the first non-directory, non-zero-byte entry in its file list. That does not guarantee selection by filename, creation time, or upload order.

Direct-query observations

I tested these objects in S3-compatible storage, separately for Parquet and ORC:

Input	Schema	Row
Older file	id BIGINT, event STRING	id=1, event='click'
Newer file	id BIGINT, event STRING, campaign STRING	id=2, event='purchase', campaign='winter'

For each format, I created two prefixes containing byte-identical copies: old_first/ used 000-old and 001-new, while new_first/ exchanged the names to 001-old and 000-new, with the appropriate format suffix. These are test labels, not ordering guarantees. The field sets and rows below come from each SELECT * statement's own returned column metadata and rows. DESC FUNCTION was a separate diagnostic statement; it does not pin the selected file for a later SELECT.

With an S3 resource configured for connection settings:

DESC FUNCTION S3(
    "uri"="s3://bucket/parquet/old_first/*.parquet",
    "format"="parquet", "resource"="s3_resource");

SELECT * FROM S3(
    "uri"="s3://bucket/parquet/old_first/*.parquet",
    "format"="parquet", "resource"="s3_resource")
ORDER BY id;

The other cases substitute new_first/ or the ORC path and format. ORDER BY sorts result rows only.

Direct read in this experiment	Observed fields
Ray Data with merge_schema=True	id, event, campaign
Doris TVF querying old_first/	id, event
Doris TVF querying new_first/	id, event, campaign

For both formats, each SELECT * returned two rows. In these runs, a separate SELECT of campaign failed with an unknown-column error for old_first/. The SELECT * results for new_first/ returned NULL in the older row and winter in the newer row. No CTAS or persistent target table was involved in these observations.

These results follow the documented default. I tested with Doris 4.1.5 FE/BE images. The Ray comparison used a development build containing the merged inference changes and PyArrow 23.0.1. The retained reproduction material includes fixture hashes, image digests, SQL, inferred schemas and raw outputs.

Proposed behavior

• Add a default-off TVF option, provisionally merge_schema, initially for Parquet/ORC S3/HDFS file queries.
• When enabled, inspect all eligible file schemas before column binding. Preserve the existing per-query reuse of the enumerated file set for schema discovery and execution.
• Initially merge supported scalar top-level fields by name, following Doris's name-resolution rules, with a deterministic output order. Same-named fields must have the same mapped Doris type; automatic type promotion is outside the initial scope.
• Exclude complex types, including ARRAY, MAP and STRUCT. If an eligible file schema contains an unsupported complex field with the option enabled, report an explicit unsupported-schema error instead of silently dropping the field. With the option disabled, retain existing complex-type reading behavior.
• Make fields nullable where needed and fill missing fields with typed NULLs. Keep the output schema fixed throughout execution.
• Report metadata-read or merge failures rather than silently returning a partial schema.

This adds metadata I/O during analysis, even for queries that later read few rows. Concurrency, timeout and cancellation details can be settled during design. A retained file list does not provide an atomic snapshot of an S3/HDFS directory or prevent objects from changing during execution.

Related work

The Lance multi-dataset TVF discussion in #66492 considered schema union. The subsequent review narrowed the first stage to identical schemas, as reflected in #69276. This is related design context, rather than a Parquet/ORC cross-file schema-merge implementation.

CTAS can propagate the inferred field set into its target table; that is supplementary downstream context, not the motivation for this direct-query proposal.

Questions

1. Would the community welcome this opt-in behavior for direct file queries without a declared table schema?
2. Is a scalar-only top-level field union, with matching mapped Doris types and complex schemas explicitly rejected, a suitable initial scope?
3. Is there other related work to coordinate with, or design experience from the Lance discussion that should inform this proposal?

[morningman]

thanks @jiangxt2 . This feature is reasonable, please go ahead!
But need to do some tests with many files cases to see the impact, because in Doris, this metadata fetch is done  by FE, in a single thread.