avro.patch

diff --git a/lang/c/src/avro/io.h b/lang/c/src/avro/io.h
index ffbb68dc5..5bb2ee699 100644
--- a/lang/c/src/avro/io.h
+++ b/lang/c/src/avro/io.h
@@ -109,12 +109,17 @@ int avro_file_writer_create_with_codec(const char *path,
 int avro_file_writer_create_with_codec_fp(FILE *fp, const char *path, int should_close,
 				avro_schema_t schema, avro_file_writer_t * writer,
 				const char *codec, size_t block_size);
+int avro_file_writer_create_with_codec_metadata_fp(FILE *fp, const char *path, int should_close,
+				avro_schema_t schema, avro_file_writer_t * writer,
+				const char *codec, size_t block_size, avro_value_t *meta);
 int avro_file_writer_open(const char *path, avro_file_writer_t * writer);
 int avro_file_writer_open_bs(const char *path, avro_file_writer_t * writer, size_t block_size);
 int avro_file_reader(const char *path, avro_file_reader_t * reader);
 int avro_file_reader_fp(FILE *fp, const char *path, int should_close,
 			avro_file_reader_t * reader);
 
+int avro_file_reader_json_schema(const char *file_path, const char **json_schema);
+
 avro_schema_t
 avro_file_reader_get_writer_schema(avro_file_reader_t reader);
 
diff --git a/lang/c/src/datafile.c b/lang/c/src/datafile.c
index c9d4dfeb6..dcb6f50cc 100644
--- a/lang/c/src/datafile.c
+++ b/lang/c/src/datafile.c
@@ -73,7 +73,7 @@ static int write_sync(avro_file_writer_t w)
 	return avro_write(w->writer, w->sync, sizeof(w->sync));
 }
 
-static int write_header(avro_file_writer_t w)
+static int write_header(avro_file_writer_t w, avro_value_t *extra_meta)
 {
 	int rval;
 	uint8_t version = 1;
@@ -82,27 +82,65 @@ static int write_header(avro_file_writer_t w)
 	const avro_encoding_t *enc = &avro_binary_encoding;
 	int64_t schema_len;
 
+	const char *json_schema = NULL;
+	size_t json_schema_len = 0;
+  avro_value_t properties = {0};
+  size_t properties_count = 0;
+	if (extra_meta != NULL)
+	{
+		avro_value_t meta_schema = {0};
+		check(rval, avro_value_get_by_name(extra_meta, "avroschema", &meta_schema, NULL));
+		check(rval, avro_value_get_string(&meta_schema, &json_schema, &json_schema_len));
+
+    /* optional map of extra string entries for the file metadata */
+    if (avro_value_get_by_name(extra_meta, "properties", &properties, NULL) == 0)
+      check(rval, avro_value_get_size(&properties, &properties_count));
+	}
+
 	/* Generate random sync */
 	generate_sync(w);
 
 	check(rval, avro_write(w->writer, "Obj", 3));
 	check(rval, avro_write(w->writer, &version, 1));
 
-	check(rval, enc->write_long(w->writer, 2));
+	check(rval, enc->write_long(w->writer, 2 + properties_count));
 	check(rval, enc->write_string(w->writer, "avro.codec"));
 	check(rval, enc->write_bytes(w->writer, w->codec->name, strlen(w->codec->name)));
 	check(rval, enc->write_string(w->writer, "avro.schema"));
 	schema_writer =
 	    avro_writer_memory(&w->schema_buf[0], sizeof(w->schema_buf));
-	rval = avro_schema_to_json(w->writers_schema, schema_writer);
+
+	if (json_schema) {
+		rval = avro_write(schema_writer, (char *)json_schema, strlen(json_schema));
+	}
+	else {
+		rval = avro_schema_to_json(w->writers_schema, schema_writer);
+	}
+
 	if (rval) {
 		avro_writer_free(schema_writer);
 		return rval;
 	}
+
 	schema_len = avro_writer_tell(schema_writer);
 	avro_writer_free(schema_writer);
 	check(rval,
 	      enc->write_bytes(w->writer, w->schema_buf, schema_len));
+
+	for (size_t i = 0; i < properties_count; i++) {
+		avro_value_t value;
+		const char *key = NULL;
+		const char *str = NULL;
+		size_t str_len = 0;
+
+		check(rval, avro_value_get_by_index(&properties, i, &value, &key));
+		check(rval, avro_value_get_string(&value, &str, &str_len));
+
+		/* str_len counts the NUL terminator, which is not part of the value */
+		check(rval, enc->write_string(w->writer, key));
+		check(rval, enc->write_bytes(w->writer, str, str_len - 1));
+	}
+
 	check(rval, enc->write_long(w->writer, 0));
 	return write_sync(w);
 }
@@ -140,7 +178,7 @@ file_writer_init_fp(FILE *fp, const char *path, int should_close, const char *mo
 #endif
 
 static int
-file_writer_create(FILE *fp, const char *path, int should_close, avro_schema_t schema, avro_file_writer_t w, size_t block_size)
+file_writer_create(FILE *fp, const char *path, int should_close, avro_schema_t schema, avro_file_writer_t w, size_t block_size, avro_value_t *extra_meta)
 {
 	int rval;
 
@@ -169,7 +207,7 @@ file_writer_create(FILE *fp, const char *path, int should_close, avro_schema_t s
 	}
 
 	w->writers_schema = avro_schema_incref(schema);
-	return write_header(w);
+	return write_header(w, extra_meta);
 }
 
 int
@@ -196,6 +234,15 @@ int avro_file_writer_create_with_codec(const char *path,
 int avro_file_writer_create_with_codec_fp(FILE *fp, const char *path, int should_close,
 			avro_schema_t schema, avro_file_writer_t * writer,
 			const char *codec, size_t block_size)
+{
+	return avro_file_writer_create_with_codec_metadata_fp(fp, path, should_close, schema, writer, codec,
+														  block_size, NULL);
+}
+
+
+int avro_file_writer_create_with_codec_metadata_fp(FILE *fp, const char *path, int should_close,
+			avro_schema_t schema, avro_file_writer_t * writer,
+			const char *codec, size_t block_size, avro_value_t *extra_meta)
 {
 	avro_file_writer_t w;
 	int rval;
@@ -226,7 +273,7 @@ int avro_file_writer_create_with_codec_fp(FILE *fp, const char *path, int should
 		avro_freet(struct avro_file_writer_t_, w);
 		return rval;
 	}
-	rval = file_writer_create(fp, path, should_close, schema, w, block_size);
+	rval = file_writer_create(fp, path, should_close, schema, w, block_size, extra_meta);
 	if (rval) {
 		avro_codec_reset(w->codec);
 		avro_freet(struct avro_codec_t_, w->codec);
@@ -452,6 +499,36 @@ static int file_read_block_count(avro_file_reader_t r)
 	check_prefix(rval, enc->read_long(r->reader, &len),
 		     "Cannot read file block size: ");
 
+	/*
+	 * Both counts come straight off the wire as zigzag varints, so a corrupt
+	 * or misaligned file can decode them to anything, including negatives.
+	 * Neither is meaningful below zero, and letting them through does real
+	 * damage:
+	 *
+	 *   - a negative len reaches avro_malloc(len) as a size_t, which is a
+	 *     nonsensically huge allocation request (embedders that route
+	 *     avro_malloc to their own allocator see it as such);
+	 *   - or, when current_blockdata is already allocated, a negative len is
+	 *     not greater than current_blocklen, so no buffer is resized, the
+	 *     "len > 0" read and decode below are both skipped, and block_reader
+	 *     is re-pointed at the *previous* block's decompressed bytes. The
+	 *     caller then happily re-reads stale records, silently returning data
+	 *     that is not in the file;
+	 *   - a negative blocks_total makes blocks_read == blocks_total never
+	 *     true, so the sync marker is never checked again.
+	 *
+	 * Fail the read instead, so a damaged file is reported as damaged.
+	 */
+	if (r->blocks_total < 0) {
+		avro_set_error("Invalid file block count %" PRId64, r->blocks_total);
+		return EILSEQ;
+	}
+
+	if (len < 0) {
+		avro_set_error("Invalid file block size %" PRId64, len);
+		return EILSEQ;
+	}
+
 	if (r->current_blockdata && len > r->current_blocklen) {
 		r->current_blockdata = (char *) avro_realloc(r->current_blockdata, r->current_blocklen, len);
 		r->current_blocklen = len;
@@ -466,6 +543,27 @@ static int file_read_block_count(avro_file_reader_t r)
 
 		check_prefix(rval, avro_codec_decode(r->codec, r->current_blockdata, len),
 			     "Cannot decode file block: ");
+	} else {
+		/*
+		 * Nothing was read, so nothing was decoded, and the codec still holds
+		 * the *previous* block's decompressed bytes. Handing those to
+		 * block_reader below would satisfy this block's record count out of
+		 * stale data, so the caller silently receives records that are not in
+		 * the file. Report an empty block instead, and a count that cannot be
+		 * satisfied from it fails the read.
+		 *
+		 * Only used_size may be cleared here. block_data is owned by codecs
+		 * that allocate an output buffer -- deflate frees it in its reset --
+		 * and it aliases current_blockdata for the null codec, so clearing the
+		 * pointer would either leak or double free.
+		 *
+		 * A zero length is not rejected outright because it is representable in
+		 * a well-formed file: a record schema with no fields encodes to zero
+		 * bytes per record, so with the null codec a block of N such records
+		 * legitimately carries a count of N and a size of 0, and libavro's own
+		 * writer emits exactly that.
+		 */
+		r->codec->used_size = 0;
 	}
 
 	avro_reader_memory_set_source(r->block_reader, (const char *) r->codec->block_data, r->codec->used_size);
@@ -541,6 +639,93 @@ int avro_file_reader_fp(FILE *fp, const char *path, int should_close,
 	return 0;
 }
 
+int
+avro_file_reader_json_schema(const char *file_path, const char **json_schema)
+{
+	FILE *file = fopen(file_path, "rb");
+	if (!file) {
+		avro_set_error("Error opening file: %s",
+			       strerror(errno));
+		return errno;
+	}
+
+	avro_reader_t reader = avro_reader_file(file);
+
+	int rval;
+
+	char		magic[4] = {0};
+
+	/* read magic footer */
+	check(rval, avro_read(reader, magic, sizeof(magic)));
+
+	if (magic[0] != 'O' || magic[1] != 'b' || magic[2] != 'j'
+		|| magic[3] != 1)
+	{
+		avro_reader_free(reader);
+		avro_set_error("Incorrect Avro container file magic number");
+		return 1;
+	}
+
+	/* each value is bytes */
+	avro_schema_t meta_values_schema = avro_schema_bytes();
+
+	/* metadata is map */
+	avro_schema_t meta_schema = avro_schema_map(meta_values_schema);
+
+	/* prepare avro interface for the schema */
+	avro_value_iface_t *meta_iface = avro_generic_class_from_schema(meta_schema);
+
+	if (meta_iface == NULL)
+	{
+		avro_reader_free(reader);
+		avro_set_error("Cannot create metadata interface");
+		return 1;
+	}
+
+	/* read avro metadata */
+	avro_value_t meta;
+
+	if (avro_generic_value_new(meta_iface, &meta) != 0)
+	{
+		avro_reader_free(reader);
+		avro_set_error("Cannot create metadata value");
+		return 1;
+	}
+
+	if (avro_value_read(reader, &meta) != 0)
+	{
+		avro_reader_free(reader);
+		avro_set_error("Cannot read file header");
+		return 1;
+	}
+
+	/* read "avro.schema" from the metadata */
+	avro_value_t schema_bytes;
+
+	if (avro_value_get_by_name(&meta, "avro.schema", &schema_bytes, NULL) != 0)
+	{
+		avro_reader_free(reader);
+		avro_set_error("File header doesn't contain a schema");
+		return 1;
+	}
+
+	const void *p = NULL;
+	size_t		len = 0;
+
+	avro_value_get_bytes(&schema_bytes, &p, &len);
+
+	char *schema = avro_malloc(len + 1);
+
+	memcpy((void *) schema, p, len);
+	schema[len] = '\0';
+
+	*json_schema = schema;
+
+	avro_reader_free(reader);
+
+	return 0;
+}
+
 int avro_file_reader(const char *path, avro_file_reader_t * reader)
 {
 	FILE *fp;
diff --git a/lang/c/src/schema.c b/lang/c/src/schema.c
index a4d8e9f89..cddea00cc 100644
--- a/lang/c/src/schema.c
+++ b/lang/c/src/schema.c
@@ -56,7 +56,7 @@ static int is_avro_id(const char *name)
 		}
 		for (i = 0; i < len; i++) {
 			if (!(isalpha(name[i])
-			      || name[i] == '_' || (i && isdigit(name[i])))) {
+			      || name[i] == '_' || name[i] == '.' || (i && isdigit(name[i])))) {
 				return 0;
 			}
 		}