| 125 | } |
| 126 | |
| 127 | Status HdfsAvroScanner::ParseMetadata() { |
| 128 | header_->is_compressed = false; |
| 129 | header_->compression_type = THdfsCompression::NONE; |
| 130 | |
| 131 | int64_t num_entries; |
| 132 | RETURN_IF_FALSE(stream_->ReadZLong(&num_entries, &parse_status_)); |
| 133 | if (num_entries < 1) { |
| 134 | return Status(TErrorCode::AVRO_INVALID_METADATA_COUNT, stream_->filename(), |
| 135 | num_entries, stream_->file_offset()); |
| 136 | } |
| 137 | |
| 138 | while (num_entries != 0) { |
| 139 | DCHECK_GT(num_entries, 0); |
| 140 | for (int i = 0; i < num_entries; ++i) { |
| 141 | // Decode Avro string-type key |
| 142 | string key; |
| 143 | uint8_t* key_buf; |
| 144 | int64_t key_len; |
| 145 | RETURN_IF_FALSE(stream_->ReadZLong(&key_len, &parse_status_)); |
| 146 | if (key_len < 0) { |
| 147 | return Status(TErrorCode::AVRO_INVALID_LENGTH, stream_->filename(), key_len, |
| 148 | stream_->file_offset()); |
| 149 | } |
| 150 | RETURN_IF_FALSE(stream_->ReadBytes(key_len, &key_buf, &parse_status_)); |
| 151 | key = string(reinterpret_cast<char*>(key_buf), key_len); |
| 152 | |
| 153 | // Decode Avro bytes-type value |
| 154 | uint8_t* value; |
| 155 | int64_t value_len; |
| 156 | RETURN_IF_FALSE(stream_->ReadZLong(&value_len, &parse_status_)); |
| 157 | if (value_len < 0) { |
| 158 | return Status(TErrorCode::AVRO_INVALID_LENGTH, stream_->filename(), value_len, |
| 159 | stream_->file_offset()); |
| 160 | } |
| 161 | RETURN_IF_FALSE(stream_->ReadBytes(value_len, &value, &parse_status_)); |
| 162 | |
| 163 | if (key == AVRO_SCHEMA_KEY) { |
| 164 | avro_schema_t raw_file_schema; |
| 165 | int error = avro_schema_from_json_length( |
| 166 | reinterpret_cast<char*>(value), value_len, &raw_file_schema); |
| 167 | if (error != 0) { |
| 168 | stringstream ss; |
| 169 | ss << "Failed to parse file schema: " << avro_strerror(); |
| 170 | return Status(ss.str()); |
| 171 | } |
| 172 | AvroSchemaElement* file_schema = avro_header_->schema.get(); |
| 173 | RETURN_IF_ERROR(AvroSchemaElement::ConvertSchema(raw_file_schema, file_schema)); |
| 174 | |
| 175 | RETURN_IF_ERROR(ResolveSchemas(scan_node_->avro_schema(), file_schema)); |
| 176 | |
| 177 | // We currently codegen a function only for the table schema. If this file's |
| 178 | // schema is different from the table schema, don't use the codegen'd function and |
| 179 | // use the interpreted path instead. |
| 180 | avro_header_->use_codegend_decode_avro_data = avro_schema_equal( |
| 181 | scan_node_->avro_schema().schema, file_schema->schema); |
| 182 | |
| 183 | } else if (key == AVRO_CODEC_KEY) { |
| 184 | string avro_codec(reinterpret_cast<char*>(value), value_len); |