[adapters] Iceberg connector reads only columns declared in the SQL table#6638
Conversation
… table Replace the `select *` snapshot queries with an explicit column list: the intersection of the Iceberg snapshot schema and the SQL table declaration, mirroring the Delta connector's `used_column_list`. Undeclared table columns previously wasted IO and could fail deserialization for column types Feldera does not support. The connector also honors the table-level `skip_unused_columns` property: a declared-but-unused column is skipped when it is nullable or has a default and `snapshot_filter` does not reference it. The property check lives in `Relation::skip_unused_columns`, used by both the Delta and Iceberg connectors. Share `ColumnNameSet` and `quote_sql_identifier` between the Delta and Iceberg connectors by moving them to feldera-adapterlib. Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
swanandx
left a comment
There was a problem hiding this comment.
PR LGTM! just have one suggestion
Approving anyways to unblock. thanks!
| fn used_sql_columns(schema: &Relation, config: &IcebergReaderConfig) -> ColumnNameSet { | ||
| let skip = schema.skip_unused_columns(); | ||
| let config_referenced = config_referenced_columns(config); | ||
| ColumnNameSet::from_names( | ||
| schema | ||
| .fields | ||
| .iter() | ||
| .filter(|f| !skip || !can_skip_column(f, &config_referenced)) | ||
| .map(|f| f.name.name()), | ||
| ) | ||
| } | ||
|
|
||
| /// Compute the subset of columns in the Iceberg table snapshot schema that | ||
| /// occur in the SQL table declaration (minus columns removed by | ||
| /// [`used_sql_columns`]), preserving the table schema's column order. | ||
| /// | ||
| /// Snapshot queries select this subset instead of `*`, so the connector never | ||
| /// reads table columns the pipeline does not ingest. Besides being wasteful, | ||
| /// reading such columns can fail when their Arrow type is not supported by | ||
| /// the Feldera deserializer. | ||
| /// | ||
| /// Iceberg schemas carry no case-sensitivity information, so column names are | ||
| /// matched in lowercased form (see [`ColumnNameSet`]). | ||
| fn used_columns( | ||
| table_schema: &ArrowSchema, | ||
| schema: &Relation, | ||
| config: &IcebergReaderConfig, | ||
| ) -> Vec<String> { | ||
| let used = used_sql_columns(schema, config); | ||
| table_schema | ||
| .fields() | ||
| .iter() | ||
| .filter(|f| used.contains(f.name())) | ||
| .map(|f| f.name().to_string()) | ||
| .collect() | ||
| } | ||
|
|
||
| /// Quoted, comma-separated column list for `select {} from snapshot` queries. | ||
| fn used_column_list(columns: &[String]) -> String { | ||
| columns | ||
| .iter() | ||
| .map(quote_sql_identifier) | ||
| .collect::<Vec<_>>() | ||
| .join(", ") | ||
| } |
There was a problem hiding this comment.
we should make declare them as methods of IcebergInputEndpointInner
where used_sql_columns: OnceLock<ColumnNameSet>andconfig_referenced_columns: OnceLock` are fields on that struct.
so we don't need to construct ColumnNameSet everytime
maybe this is fine for now ( as there is just snapshot mode ) but i would say let's stay consistent for now, and get rid of it if it isn't required later, wdyt?
There was a problem hiding this comment.
Can't it be just computed during initialization. Either way, computing them once makes sense to me. I suppose we want to do this in both iceberg and delta connectors.
There was a problem hiding this comment.
delta already does declare them as i mentioned above, computing them just once. So we need to do this just for iceberg as quoted here
| @@ -241,11 +268,16 @@ fn iceberg_localfs_input_test( | |||
| let script_path = "../iceberg/src/test/create_test_table_s3.py"; | |||
There was a problem hiding this comment.
Nit: the changelog checkbox is unchecked. This is a user-visible behavior change (Iceberg connector now reads only declared columns instead of *, and honors skip_unused_columns). Worth a one-liner under Unreleased.
Replace the
select *snapshot queries with an explicit column list: the intersection of the Iceberg snapshot schema and the SQL table declaration, mirroring the Delta connector'sused_column_list. Undeclared table columns previously wasted IO and could fail deserialization for column types Feldera does not support.The connector also honors the table-level
skip_unused_columnsproperty: a declared-but-unused column is skipped when it is nullable or has a default andsnapshot_filterdoes not reference it. The property check lives inRelation::skip_unused_columns, used by both the Delta and Iceberg connectors.Share
ColumnNameSetandquote_sql_identifierbetween the Delta and Iceberg connectors by moving them to feldera-adapterlib.Describe Manual Test Plan
Checklist
Breaking Changes?
Mark if you think the answer is yes for any of these components:
Describe Incompatible Changes