Generic.cc 7.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228
  1. /**
  2. * Licensed to the Apache Software Foundation (ASF) under one
  3. * or more contributor license agreements. See the NOTICE file
  4. * distributed with this work for additional information
  5. * regarding copyright ownership. The ASF licenses this file
  6. * to you under the Apache License, Version 2.0 (the
  7. * "License"); you may not use this file except in compliance
  8. * with the License. You may obtain a copy of the License at
  9. *
  10. * https://www.apache.org/licenses/LICENSE-2.0
  11. *
  12. * Unless required by applicable law or agreed to in writing, software
  13. * distributed under the License is distributed on an "AS IS" BASIS,
  14. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. * See the License for the specific language governing permissions and
  16. * limitations under the License.
  17. */
  18. #include "Generic.hh"
  19. #include <utility>
  20. namespace avro {
  21. using std::ostringstream;
  22. using std::string;
  23. using std::vector;
  24. typedef vector<uint8_t> bytes;
  25. void GenericContainer::assertType(const NodePtr &schema, Type type) {
  26. if (schema->type() != type) {
  27. throw Exception("Schema type {} expected {}", schema->type(), type);
  28. }
  29. }
  30. GenericReader::GenericReader(ValidSchema s, const DecoderPtr &decoder) : schema_(std::move(s)), isResolving_(dynamic_cast<ResolvingDecoder *>(&(*decoder)) != nullptr),
  31. decoder_(decoder) {
  32. }
  33. GenericReader::GenericReader(const ValidSchema &writerSchema,
  34. const ValidSchema &readerSchema, const DecoderPtr &decoder) : schema_(readerSchema),
  35. isResolving_(true),
  36. decoder_(resolvingDecoder(writerSchema, readerSchema, decoder)) {
  37. }
  38. void GenericReader::read(GenericDatum &datum) const {
  39. datum = GenericDatum(schema_.root());
  40. read(datum, *decoder_, isResolving_);
  41. }
  42. void GenericReader::read(GenericDatum &datum, Decoder &d, bool isResolving) {
  43. if (datum.isUnion()) {
  44. datum.selectBranch(d.decodeUnionIndex());
  45. }
  46. switch (datum.type()) {
  47. case AVRO_NULL:
  48. d.decodeNull();
  49. break;
  50. case AVRO_BOOL:
  51. datum.value<bool>() = d.decodeBool();
  52. break;
  53. case AVRO_INT:
  54. datum.value<int32_t>() = d.decodeInt();
  55. break;
  56. case AVRO_LONG:
  57. datum.value<int64_t>() = d.decodeLong();
  58. break;
  59. case AVRO_FLOAT:
  60. datum.value<float>() = d.decodeFloat();
  61. break;
  62. case AVRO_DOUBLE:
  63. datum.value<double>() = d.decodeDouble();
  64. break;
  65. case AVRO_STRING:
  66. d.decodeString(datum.value<string>());
  67. break;
  68. case AVRO_BYTES:
  69. d.decodeBytes(datum.value<bytes>());
  70. break;
  71. case AVRO_FIXED: {
  72. auto &f = datum.value<GenericFixed>();
  73. d.decodeFixed(f.schema()->fixedSize(), f.value());
  74. } break;
  75. case AVRO_RECORD: {
  76. auto &r = datum.value<GenericRecord>();
  77. size_t c = r.schema()->leaves();
  78. if (isResolving) {
  79. std::vector<size_t> fo =
  80. static_cast<ResolvingDecoder &>(d).fieldOrder();
  81. for (size_t i = 0; i < c; ++i) {
  82. read(r.fieldAt(fo[i]), d, isResolving);
  83. }
  84. } else {
  85. for (size_t i = 0; i < c; ++i) {
  86. read(r.fieldAt(i), d, isResolving);
  87. }
  88. }
  89. } break;
  90. case AVRO_ENUM:
  91. datum.value<GenericEnum>().set(d.decodeEnum());
  92. break;
  93. case AVRO_ARRAY: {
  94. auto &v = datum.value<GenericArray>();
  95. vector<GenericDatum> &r = v.value();
  96. const NodePtr &nn = v.schema()->leafAt(0);
  97. r.resize(0);
  98. size_t start = 0;
  99. for (size_t m = d.arrayStart(); m != 0; m = d.arrayNext()) {
  100. r.resize(r.size() + m);
  101. for (; start < r.size(); ++start) {
  102. r[start] = GenericDatum(nn);
  103. read(r[start], d, isResolving);
  104. }
  105. }
  106. } break;
  107. case AVRO_MAP: {
  108. auto &v = datum.value<GenericMap>();
  109. GenericMap::Value &r = v.value();
  110. const NodePtr &nn = v.schema()->leafAt(1);
  111. r.resize(0);
  112. size_t start = 0;
  113. for (size_t m = d.mapStart(); m != 0; m = d.mapNext()) {
  114. r.resize(r.size() + m);
  115. for (; start < r.size(); ++start) {
  116. d.decodeString(r[start].first);
  117. r[start].second = GenericDatum(nn);
  118. read(r[start].second, d, isResolving);
  119. }
  120. }
  121. } break;
  122. default:
  123. throw Exception("Unknown schema type {}", datum.type());
  124. }
  125. }
  126. void GenericReader::read(Decoder &d, GenericDatum &g, const ValidSchema &s) {
  127. g = GenericDatum(s);
  128. read(d, g);
  129. }
  130. void GenericReader::read(Decoder &d, GenericDatum &g) {
  131. read(g, d, dynamic_cast<ResolvingDecoder *>(&d) != nullptr);
  132. }
  133. GenericWriter::GenericWriter(ValidSchema s, EncoderPtr encoder) : schema_(std::move(s)), encoder_(std::move(encoder)) {
  134. }
  135. void GenericWriter::write(const GenericDatum &datum) const {
  136. write(datum, *encoder_);
  137. }
  138. void GenericWriter::write(const GenericDatum &datum, Encoder &e) {
  139. if (datum.isUnion()) {
  140. e.encodeUnionIndex(datum.unionBranch());
  141. }
  142. switch (datum.type()) {
  143. case AVRO_NULL:
  144. e.encodeNull();
  145. break;
  146. case AVRO_BOOL:
  147. e.encodeBool(datum.value<bool>());
  148. break;
  149. case AVRO_INT:
  150. e.encodeInt(datum.value<int32_t>());
  151. break;
  152. case AVRO_LONG:
  153. e.encodeLong(datum.value<int64_t>());
  154. break;
  155. case AVRO_FLOAT:
  156. e.encodeFloat(datum.value<float>());
  157. break;
  158. case AVRO_DOUBLE:
  159. e.encodeDouble(datum.value<double>());
  160. break;
  161. case AVRO_STRING:
  162. e.encodeString(datum.value<string>());
  163. break;
  164. case AVRO_BYTES:
  165. e.encodeBytes(datum.value<bytes>());
  166. break;
  167. case AVRO_FIXED:
  168. e.encodeFixed(datum.value<GenericFixed>().value());
  169. break;
  170. case AVRO_RECORD: {
  171. const auto &r = datum.value<GenericRecord>();
  172. size_t c = r.schema()->leaves();
  173. for (size_t i = 0; i < c; ++i) {
  174. write(r.fieldAt(i), e);
  175. }
  176. } break;
  177. case AVRO_ENUM:
  178. e.encodeEnum(datum.value<GenericEnum>().value());
  179. break;
  180. case AVRO_ARRAY: {
  181. const GenericArray::Value &r = datum.value<GenericArray>().value();
  182. e.arrayStart();
  183. if (!r.empty()) {
  184. e.setItemCount(r.size());
  185. for (const auto &it : r) {
  186. e.startItem();
  187. write(it, e);
  188. }
  189. }
  190. e.arrayEnd();
  191. } break;
  192. case AVRO_MAP: {
  193. const GenericMap::Value &r = datum.value<GenericMap>().value();
  194. e.mapStart();
  195. if (!r.empty()) {
  196. e.setItemCount(r.size());
  197. for (const auto &it : r) {
  198. e.startItem();
  199. e.encodeString(it.first);
  200. write(it.second, e);
  201. }
  202. }
  203. e.mapEnd();
  204. } break;
  205. default:
  206. throw Exception("Unknown schema type {}", datum.type());
  207. }
  208. }
  209. void GenericWriter::write(Encoder &e, const GenericDatum &g) {
  210. write(g, e);
  211. }
  212. } // namespace avro