Conversation
|
@codex review |
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 090c16ac1f
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
d5099b8 to
ea636f0
Compare
This comment was marked as outdated.
This comment was marked as outdated.
Reuse `ISerialization` for aggregate state conversion, trim nonessential comments, and consolidate overlapping tests while preserving distinct format and Iceberg coverage. PR: #2301
Parquet and Iceberg have no aggregate-state type, so neither can describe an `AggregateFunction` or `SimpleAggregateFunction` column with its schema alone. This adds a ClickHouse-only annotation next to the data - the `clickhouse.column_types` key of a parquet file's footer, and a `clickhouse.type` key on an Iceberg schema field - naming the ClickHouse type so it can be rebuilt on read. An `AggregateFunction` state is stored as an opaque `BYTE_ARRAY` / `binary` holding the bytes the `-State` combinator produces; a `SimpleAggregateFunction(f, T)` is stored as an ordinary value of `T`. Other query engines ignore the key and see a plain binary or `T`-typed column. The recorded name always carries the state version, which pins the serialized layout: `getName` drops a zero version, so a versioned function pinned to version 0 would otherwise be rebuilt with the default version and its bytes misread. `getNameForAnnotation` keeps it. Two experimental settings gate the feature, both off by default: `allow_experimental_aggregate_function_states_in_parquet` for writing such a column and for reconstructing one during parquet schema inference, and `allow_experimental_aggregate_function_states_in_iceberg` for `CREATE TABLE` and for honouring a `clickhouse.type` key that names an `AggregateFunction`. Honouring the annotation lets the file or the table metadata, rather than the query, choose the deserializer the stored bytes are handed to, so a refused annotation is rejected rather than read as `String` - reading states as strings would be a wrong result, not an error. `SimpleAggregateFunction` needs no opt-in on read, holding ordinary values with no state deserializer involved. The annotation is checked against the type the schema derives on its own before it is honoured, so a stale or crafted one cannot re-type a column to anything of the same nesting shape. The parquet reader checks strictly, requiring a type its own writer maps to what the file holds; Iceberg checks the nesting structure, which is all its type system allows. The Iceberg gate is read from the context of the query that parses the schema rather than from one captured at `ATTACH`, so a reading query can opt in at all; `IcebergSchemaProcessor::addIcebergTableSchema` publishes a schema-id only once every field has parsed and drops what it wrote on failure, so a refused read can be retried with the setting on. The metadata prefetcher asks for the schema-id with `SchemaParsing::Skip`, since it has no query to carry the gate and must not publish a schema that later queries would be served. The parquet schema cache is keyed by both settings, so a schema inferred with a gate open is not served to a query that has it closed. `ALTER TABLE ... EXPORT PART` and `EXPORT PARTITION` from an `AggregatingMergeTree` table into an Iceberg table carry both gates: `EXPORT PARTITION` records them in its ZooKeeper manifest, so every replica executing the task applies the values the `ALTER` gave instead of its own profile, and the destination is resolved under those settings. A manifest without the fields, written before they existed, reads them as closed. Iceberg records no bounds for aggregate states, whose extremes cannot be compared, and parquet min/max statistics over serialized states are never used for pruning. Adds `04673_parquet_aggregate_function_state`, integration tests for the export paths and for round-tripping states through Spark, and unit tests for the annotation matching, the name parsing, the schema processor and the metadata generator. Documents the feature in `Parquet.md` and `iceberg.md`. PR: #2301
de5caea to
322a47b
Compare
…ion-states-in-parquet-iceberg
Reuse `removeLowCardinalityAndNullable`, inline the one-use schema helper, and remove redundant footer state and comments. Share metadata-test lookup code while preserving all regression scenarios. Validated with a Debug build, 103 targeted unit tests, and the Parquet aggregate-state regression matching its reference output. Related: #2301
…et-iceberg' of github.com:Altinity/ClickHouse into feature/antalya-26.6/aggregate-function-states-in-parquet-iceberg
…ion-states-in-parquet-iceberg
arthurpassos
left a comment
There was a problem hiding this comment.
I have reviewed the docs and the parquet implementation, looks ok so far. I'll review the iceberg impl and the tests tomorrow. Few comments below
| namespace parq = parquet::format; | ||
|
|
||
| /// Maps top-level columns to ClickHouse types not represented by the Parquet schema. | ||
| constexpr const char * clickhouse_column_types_key = "clickhouse.column_types"; |
There was a problem hiding this comment.
hm.. it would be funny if upstream one day decided to use the very same kvp for a different implementation, but I suppose it is extremely unlikely
There was a problem hiding this comment.
well, yes, there is always a chance of something like this. but I do not see any other option actually.
There was a problem hiding this comment.
I think I just take this PR into upstream later, there is (mostly) nothing antalya-specific in it
| Poco::JSON::Parser parser; | ||
| const auto object = parser.parse(kv.value).extract<Poco::JSON::Object::Ptr>(); | ||
| for (const auto & name : object->getNames()) | ||
| clickhouse_column_type_names[name] = object->getValue<String>(name); |
There was a problem hiding this comment.
A json key value pair inside a parquet key value parquet, that's funny, but probably the best way
| std::unordered_map<String, GeoColumnMetadata> geo_columns; | ||
|
|
||
| /// Type names from the `clickhouse.column_types` footer metadata. | ||
| std::unordered_map<String, String> clickhouse_column_type_names; |
There was a problem hiding this comment.
I would make the name a bit more descriptive by adding the custom keyword
There was a problem hiding this comment.
agree. claude suggested annotated_column_type_names, which IMO looks even better
|
|
||
| DataTypePtr SchemaConverter::resolveAnnotatedType(const String & column_name, const String & type_name) const | ||
| { | ||
| ASTPtr ast; |
There was a problem hiding this comment.
Instead of parsing text into an AST, ins't it easier to simply do the below? Ofc it assumes type_name is the "full type name" the data type factory requires
auto data_type = DataTypeFactory::instance().get(type_name));
if (WhichDataType(data_type).isAggregateFunction() && !options.format.parquet.allow_aggregate_function_states)
throw
There was a problem hiding this comment.
Ok, I suppose it is not that simple because it might be a nested data type. In that case you'd need something similar to what you've implemented in needsClickHouseTypeAnnotation
| { | ||
| annotated_type = resolveAnnotatedType(col.name, it->second); | ||
|
|
||
| if (!annotatedTypeMatchesDerived(annotated_type, col.output_type, /*strict=*/ true)) |
There was a problem hiding this comment.
I feel like annotatedTypeMatchesDerived should be called from inside the resolveAnnotatedType, but it is up to you
arthurpassos
left a comment
There was a problem hiding this comment.
Reviewed the tests, looks ok. Pending part is the iceberg implementation
`SchemaConverter::resolveAnnotatedType` now takes the type derived from the parquet schema and verifies it against the annotated type with `annotatedTypeMatchesDerived`, so it only returns a validated type. Related: #2301 (comment)
…egate_function_states_in_open_formats` Replace `allow_experimental_aggregate_function_states_in_parquet` and `allow_experimental_aggregate_function_states_in_iceberg` with a single setting. Both gated the same trust decision: honouring a ClickHouse type recorded in the data, which chooses the deserializer for the stored bytes. Each check keeps its current behavior. The export task and manifest now carry one flag. Related: #2301 (comment)
Closes #2206
Changelog category (leave one):
Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):
Support aggregate function states in Parquet and Iceberg
CI/CD Options
Exclude tests:
Regression jobs to run: