| 116 | } |
| 117 | |
| 118 | pub fn read(self, data_splits: &[DataSplit]) -> crate::Result<ArrowRecordBatchStream> { |
| 119 | // Build the internal read type for thin-mode files. |
| 120 | // Physical file schema: [_SEQUENCE_NUMBER, _VALUE_KIND, all_user_cols...] |
| 121 | // We need: _SEQ + _VK + union(read_type, primary_keys) |
| 122 | let seq_field = DataField::new( |
| 123 | SEQUENCE_NUMBER_FIELD_ID, |
| 124 | SEQUENCE_NUMBER_FIELD_NAME.to_string(), |
| 125 | PaimonDataType::BigInt(BigIntType::new()), |
| 126 | ); |
| 127 | let value_kind_field = DataField::new( |
| 128 | VALUE_KIND_FIELD_ID, |
| 129 | VALUE_KIND_FIELD_NAME.to_string(), |
| 130 | PaimonDataType::TinyInt(TinyIntType::new()), |
| 131 | ); |
| 132 | |
| 133 | let key_names: std::collections::HashSet<&str> = self |
| 134 | .config |
| 135 | .primary_keys |
| 136 | .iter() |
| 137 | .map(|s| s.as_str()) |
| 138 | .collect(); |
| 139 | |
| 140 | // Collect key fields from table schema. |
| 141 | let key_fields: Vec<DataField> = self |
| 142 | .config |
| 143 | .primary_keys |
| 144 | .iter() |
| 145 | .map(|pk| { |
| 146 | self.config |
| 147 | .table_fields |
| 148 | .iter() |
| 149 | .find(|f| f.name() == pk) |
| 150 | .cloned() |
| 151 | .ok_or_else(|| Error::UnexpectedError { |
| 152 | message: format!("Primary key column '{pk}' not found in table schema"), |
| 153 | source: None, |
| 154 | }) |
| 155 | }) |
| 156 | .collect::<crate::Result<Vec<_>>>()?; |
| 157 | |
| 158 | // User columns = read_type fields + any key fields not already in read_type |
| 159 | // + any sequence fields not already included. |
| 160 | let read_type_names: std::collections::HashSet<&str> = |
| 161 | self.config.read_type.iter().map(|f| f.name()).collect(); |
| 162 | let mut user_fields: Vec<DataField> = self.config.read_type.clone(); |
| 163 | for kf in &key_fields { |
| 164 | if !read_type_names.contains(kf.name()) { |
| 165 | user_fields.push(kf.clone()); |
| 166 | } |
| 167 | } |
| 168 | // Add sequence fields if not already present. |
| 169 | for sf_name in &self.config.sequence_fields { |
| 170 | if user_fields.iter().all(|f| f.name() != sf_name.as_str()) { |
| 171 | let sf = self |
| 172 | .config |
| 173 | .table_fields |
| 174 | .iter() |
| 175 | .find(|f| f.name() == sf_name.as_str()) |